This is an automated email from the ASF dual-hosted git repository.
dsmiley pushed a commit to branch branch_10x
in repository https://gitbox.apache.org/repos/asf/solr.git
The following commit(s) were added to refs/heads/branch_10x by this push:
new 2c426f9a943 SOLR-18062: Support arbitrary Kafka properties via env and
sysprops (#4185)
2c426f9a943 is described below
commit 2c426f9a94336930c44c6af4f1760e8b76cab405
Author: Gaurav Singh <[email protected]>
AuthorDate: Mon Aug 31 09:07:04 2026 -0700
SOLR-18062: Support arbitrary Kafka properties via env and sysprops (#4185)
Co-authored-by: openworld-maker
<[email protected]>
Co-authored-by: Eric Pugh <[email protected]>
(cherry picked from commit 4551969baf1dd3b1ff2fcbf1affb0e17ae800085)
---
.../unreleased/{solr-18062.yml => SOLR-18062.yml} | 0
.../org/apache/solr/crossdc/common/ConfUtil.java | 76 +++++++++++++++-----
.../apache/solr/crossdc/common/ConfUtilTest.java | 81 +++++++++++++++++++++-
.../pages/cross-dc-replication.adoc | 4 ++
4 files changed, 144 insertions(+), 17 deletions(-)
diff --git a/changelog/unreleased/solr-18062.yml
b/changelog/unreleased/SOLR-18062.yml
similarity index 100%
rename from changelog/unreleased/solr-18062.yml
rename to changelog/unreleased/SOLR-18062.yml
diff --git
a/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/ConfUtil.java
b/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/ConfUtil.java
index 924f2be802b..ab831845096 100644
---
a/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/ConfUtil.java
+++
b/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/ConfUtil.java
@@ -22,6 +22,7 @@ import static
org.apache.solr.crossdc.common.KafkaCrossDcConf.TOPIC_NAME;
import java.io.ByteArrayInputStream;
import java.lang.invoke.MethodHandles;
import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.Locale;
import java.util.Map;
import java.util.Properties;
@@ -53,12 +54,7 @@ public class ConfUtil {
}
}
// fill in aux Kafka env with prefix
- env.forEach(
- (key, val) -> {
- if (key.startsWith(KAFKA_ENV_PREFIX)) {
- properties.put(normalizeKafkaEnvKey(key), val);
- }
- });
+ addAdditionalKafkaProperties(properties, env, System.getProperties());
// fill in from system properties
for (ConfigProperty configKey : KafkaCrossDcConf.CONFIG_PROPERTIES) {
@@ -67,14 +63,6 @@ public class ConfUtil {
properties.put(configKey.getKey(), val);
}
}
- // fill in aux Kafka system properties with prefix
- System.getProperties()
- .forEach(
- (key, val) -> {
- if (key.toString().startsWith(KAFKA_PROP_PREFIX)) {
- properties.put(normalizeKafkaSysPropKey(key.toString()), val);
- }
- });
Properties zkProps = new Properties();
if (solrClient != null) {
@@ -117,15 +105,56 @@ public class ConfUtil {
Set<String> keys = new HashSet<>(properties.keySet());
keys.forEach(
key -> {
+ Object value = properties.get(key);
if (key.startsWith(KAFKA_ENV_PREFIX)) {
- properties.put(normalizeKafkaEnvKey(key),
properties.remove(key));
+ properties.remove(key);
+ putIfNonBlankAndMissing(properties, normalizeKafkaEnvKey(key),
value);
} else if (key.startsWith(KAFKA_PROP_PREFIX)) {
- properties.put(normalizeKafkaSysPropKey(key),
properties.remove(key));
+ properties.remove(key);
+ putIfNonBlankAndMissing(properties,
normalizeKafkaSysPropKey(key), value);
}
});
}
}
+ // System properties override environment variables for pass-through Kafka
properties.
+ // Existing explicit keys in properties are preserved.
+ static void addAdditionalKafkaProperties(
+ Map<String, Object> properties, Map<String, String> env, Properties
sysProps) {
+ Set<String> envDerivedKeys = new LinkedHashSet<>();
+ env.forEach(
+ (key, val) -> {
+ if (!key.startsWith(KAFKA_ENV_PREFIX)) {
+ return;
+ }
+ String normalized = normalizeKafkaEnvKey(key);
+ if (!isValidAdditionalProperty(normalized, val)) {
+ return;
+ }
+ Object existingValue = properties.get(normalized);
+ if (isBlankValue(existingValue)) {
+ properties.put(normalized, val);
+ envDerivedKeys.add(normalized);
+ }
+ });
+
+ sysProps.forEach(
+ (key, val) -> {
+ String propKey = key.toString();
+ if (!propKey.startsWith(KAFKA_PROP_PREFIX)) {
+ return;
+ }
+ String normalized = normalizeKafkaSysPropKey(propKey);
+ if (!isValidAdditionalProperty(normalized, val)) {
+ return;
+ }
+ Object existingValue = properties.get(normalized);
+ if (isBlankValue(existingValue) ||
envDerivedKeys.contains(normalized)) {
+ properties.put(normalized, val.toString());
+ }
+ });
+ }
+
public static String normalizeKafkaEnvKey(String key) {
if (key.startsWith(KAFKA_ENV_PREFIX)) {
return
key.substring(KAFKA_ENV_PREFIX.length()).toLowerCase(Locale.ROOT).replace('_',
'.');
@@ -142,6 +171,21 @@ public class ConfUtil {
}
}
+ private static boolean isValidAdditionalProperty(String key, Object value) {
+ return key != null && !key.isBlank() && !isBlankValue(value);
+ }
+
+ private static boolean isBlankValue(Object value) {
+ return value == null || (value instanceof String && ((String)
value).isBlank());
+ }
+
+ private static void putIfNonBlankAndMissing(
+ Map<String, Object> properties, String key, Object value) {
+ if (isValidAdditionalProperty(key, value) && properties.get(key) == null) {
+ properties.put(key, value);
+ }
+ }
+
public static void verifyProperties(Map<String, Object> properties) {
if (properties.get(BOOTSTRAP_SERVERS) == null) {
log.error(
diff --git
a/solr/modules/cross-dc/src/test/org/apache/solr/crossdc/common/ConfUtilTest.java
b/solr/modules/cross-dc/src/test/org/apache/solr/crossdc/common/ConfUtilTest.java
index 53713cdfc12..6afcc35ae2a 100644
---
a/solr/modules/cross-dc/src/test/org/apache/solr/crossdc/common/ConfUtilTest.java
+++
b/solr/modules/cross-dc/src/test/org/apache/solr/crossdc/common/ConfUtilTest.java
@@ -150,7 +150,7 @@ public class ConfUtilTest extends SolrTestCaseJ4 {
public void testFillProperties_PriorityOrder() throws Exception {
Map<String, Object> properties = new HashMap<>();
- // Set up ZooKeeper with lowest priority
+ // Set up ZooKeeper with the lowest priority
Properties zkProps = new Properties();
zkProps.setProperty(KafkaCrossDcConf.BOOTSTRAP_SERVERS, "zk-kafka:9092");
zkProps.setProperty(KafkaCrossDcConf.TOPIC_NAME, "zk-topic");
@@ -347,6 +347,85 @@ public class ConfUtilTest extends SolrTestCaseJ4 {
assertEquals("zk-value", properties.get("zk.only.property"));
}
+ @Test
+ public void testAddAdditionalKafkaProperties_MapsEnvKey() {
+ Map<String, Object> properties = new HashMap<>();
+ Map<String, String> env = Map.of("SOLR_CROSSDC_KAFKA_FOO_BAR",
"env-value");
+
+ ConfUtil.addAdditionalKafkaProperties(properties, env, new Properties());
+
+ assertEquals("env-value", properties.get("foo.bar"));
+ }
+
+ @Test
+ public void testAddAdditionalKafkaProperties_MapsSysPropKey() {
+ Map<String, Object> properties = new HashMap<>();
+ Properties sysProps = new Properties();
+ sysProps.setProperty("solr.crossdc.kafka.foo.bar", "sys-value");
+
+ ConfUtil.addAdditionalKafkaProperties(properties, Map.of(), sysProps);
+
+ assertEquals("sys-value", properties.get("foo.bar"));
+ }
+
+ @Test
+ public void testAddAdditionalKafkaProperties_SysPropsOverrideEnv() {
+ Map<String, Object> properties = new HashMap<>();
+ Map<String, String> env = Map.of("SOLR_CROSSDC_KAFKA_FOO_BAR",
"env-value");
+ Properties sysProps = new Properties();
+ sysProps.setProperty("solr.crossdc.kafka.foo.bar", "sys-value");
+
+ ConfUtil.addAdditionalKafkaProperties(properties, env, sysProps);
+
+ assertEquals("sys-value", properties.get("foo.bar"));
+ }
+
+ @Test
+ public void
testAddAdditionalKafkaProperties_DoesNotOverrideExplicitProperty() {
+ Map<String, Object> properties = new HashMap<>();
+ properties.put("foo.bar", "explicit");
+ Map<String, String> env = Map.of("SOLR_CROSSDC_KAFKA_FOO_BAR",
"env-value");
+ Properties sysProps = new Properties();
+ sysProps.setProperty("solr.crossdc.kafka.foo.bar", "sys-value");
+
+ ConfUtil.addAdditionalKafkaProperties(properties, env, sysProps);
+
+ assertEquals("explicit", properties.get("foo.bar"));
+ }
+
+ @Test
+ public void
testAddAdditionalKafkaProperties_IgnoresBlankValuesAndEmptyKeys() {
+ Map<String, Object> properties = new HashMap<>();
+ Map<String, String> env = new HashMap<>();
+ env.put("SOLR_CROSSDC_KAFKA_", "ignored-empty-key");
+ env.put("SOLR_CROSSDC_KAFKA_FOO_BAR", "");
+ env.put("SOLR_CROSSDC_KAFKA_BAZ_QUX", " ");
+ Properties sysProps = new Properties();
+ sysProps.setProperty("solr.crossdc.kafka.", "ignored-empty-key");
+ sysProps.setProperty("solr.crossdc.kafka.foo.bar", "");
+ sysProps.setProperty("solr.crossdc.kafka.baz.qux", " ");
+
+ ConfUtil.addAdditionalKafkaProperties(properties, env, sysProps);
+
+ assertTrue(properties.isEmpty());
+ }
+
+ @Test
+ public void
testFillProperties_PassThroughPreservedInKafkaCrossDcConfAdditionalProperties()
{
+ Map<String, Object> properties = new HashMap<>();
+ properties.put(KafkaCrossDcConf.BOOTSTRAP_SERVERS, "localhost:9092");
+ properties.put(KafkaCrossDcConf.TOPIC_NAME, "test-topic");
+ Map<String, String> env = Map.of("SOLR_CROSSDC_KAFKA_MAX_POLL_RECORDS",
"500");
+ Properties sysProps = new Properties();
+ sysProps.setProperty("solr.crossdc.kafka.compression.type", "lz4");
+
+ ConfUtil.addAdditionalKafkaProperties(properties, env, sysProps);
+ KafkaCrossDcConf conf = new KafkaCrossDcConf(properties);
+
+ assertEquals(500, conf.getAdditionalProperties().get("max.poll.records"));
+ assertEquals("lz4",
conf.getAdditionalProperties().get("compression.type"));
+ }
+
// we can't easily modify envvars, test just the key conversion in properties
@Test
public void testUnderscoreToDotsConversion() {
diff --git
a/solr/solr-ref-guide/modules/deployment-guide/pages/cross-dc-replication.adoc
b/solr/solr-ref-guide/modules/deployment-guide/pages/cross-dc-replication.adoc
index 4a00d24fde7..6b19d64dd43 100644
---
a/solr/solr-ref-guide/modules/deployment-guide/pages/cross-dc-replication.adoc
+++
b/solr/solr-ref-guide/modules/deployment-guide/pages/cross-dc-replication.adoc
@@ -144,6 +144,10 @@ environment variables starting with `SOLR_CROSSDC_KAFKA_`
or system properties s
`solr.crossdc.kafka.`. For example, to adjust the `max.poll.records` property
specific to the Kafka client
library you can set either `SOLR_CROSSDC_KAFKA_MAX_POLL_RECORDS` environment
variable or `solr.crossdc.kafka.max.poll.records`
system property.
+Property names from environment variables are normalized by stripping the
prefix, converting to lowercase, and replacing `_` with `.`.
+Values from `solr.crossdc.kafka.` system properties override values from
`SOLR_CROSSDC_KAFKA_` environment variables when they map to the same Kafka key.
+Already-populated explicit CrossDC settings are preserved and are not
overridden by pass-through values.
+Blank values and empty normalized keys are ignored.
=== CrossDC Manager
