This is an automated email from the ASF dual-hosted git repository.
sigram pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/solr.git
The following commit(s) were added to refs/heads/main by this push:
new c43a9a56d19 SOLR-18411: CrossDC: Configure security properties in
KafkaMirroringSink initConsumer. (#4883)
c43a9a56d19 is described below
commit c43a9a56d199f9405bb0404822ddfd7c1391a936
Author: Andrzej Białecki <[email protected]>
AuthorDate: Tue Sep 8 18:53:24 2026 +0200
SOLR-18411: CrossDC: Configure security properties in KafkaMirroringSink
initConsumer. (#4883)
---
changelog/unreleased/solr-18411.yml | 8 ++
.../solr/crossdc/common/KafkaMirroringSink.java | 3 +
.../crossdc/common/KafkaMirroringSinkTest.java | 101 +++++++++++++++++++++
3 files changed, 112 insertions(+)
diff --git a/changelog/unreleased/solr-18411.yml
b/changelog/unreleased/solr-18411.yml
new file mode 100644
index 00000000000..9c0db562f88
--- /dev/null
+++ b/changelog/unreleased/solr-18411.yml
@@ -0,0 +1,8 @@
+title: CrossDC KafkaProducerSink's KafkaConsumer gets no security config
+
+type: fixed
+authors:
+ - name: Andrzej Bialecki
+links:
+ - name: SOLR-18411
+ url: https://issues.apache.org/jira/browse/SOLR-18411
diff --git
a/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/KafkaMirroringSink.java
b/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/KafkaMirroringSink.java
index 8935160937f..bd7492d1947 100644
---
a/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/KafkaMirroringSink.java
+++
b/solr/modules/cross-dc/src/java/org/apache/solr/crossdc/common/KafkaMirroringSink.java
@@ -214,6 +214,9 @@ public class KafkaMirroringSink implements
RequestMirroringSink, Closeable {
conf.getInt(KafkaCrossDcConf.MAX_PARTITION_FETCH_BYTES));
kafkaConsumerProperties.put(
ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,
conf.getInt(KafkaCrossDcConf.REQUEST_TIMEOUT_MS));
+
+ KafkaCrossDcConf.addSecurityProps(conf, kafkaConsumerProperties);
+
kafkaConsumerProperties.putAll(conf.getAdditionalProperties());
return new KafkaConsumer<>(
diff --git
a/solr/modules/cross-dc/src/test/org/apache/solr/crossdc/common/KafkaMirroringSinkTest.java
b/solr/modules/cross-dc/src/test/org/apache/solr/crossdc/common/KafkaMirroringSinkTest.java
new file mode 100644
index 00000000000..bafded545cc
--- /dev/null
+++
b/solr/modules/cross-dc/src/test/org/apache/solr/crossdc/common/KafkaMirroringSinkTest.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.crossdc.common;
+
+import static org.apache.solr.SolrTestCaseJ4.assumeWorkingMockito;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.when;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.common.config.SslConfigs;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+public class KafkaMirroringSinkTest {
+
+ @BeforeClass
+ public static void ensureWorkingMockito() {
+ assumeWorkingMockito();
+ }
+
+ @Test
+ @SuppressWarnings({"rawtypes", "try"})
+ public void testSecurityPropsApplied() throws Exception {
+ Map<String, Object> properties = new HashMap<>();
+ properties.put(KafkaCrossDcConf.TOPIC_NAME, "test-topic");
+ properties.put(KafkaCrossDcConf.BOOTSTRAP_SERVERS, "localhost:9092");
+ properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
+ properties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
"/tmp/truststore.jks");
+ properties.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "changeit");
+
+ KafkaCrossDcConf conf = new KafkaCrossDcConf(properties);
+
+ // KafkaProducer/KafkaConsumer construction is mocked out so no real
client or network is used
+ // need to suppress "try" warning to ignore args that are unused on the
surface
+ try (MockedConstruction<KafkaProducer> ignoredProducer =
mockConstruction(KafkaProducer.class);
+ MockedConstruction<KafkaConsumer> mockedConsumer =
+ mockConstruction(
+ KafkaConsumer.class,
+ // need this topic, too
+ (mock, context) ->
+ when(mock.listTopics()).thenReturn(Map.of("test-topic",
List.of())));
+ MockedStatic<KafkaCrossDcConf> mockedStatic =
+ mockStatic(KafkaCrossDcConf.class, CALLS_REAL_METHODS)) {
+
+ // collect all newly created properties
+ ArgumentCaptor<Properties> propsCaptor =
ArgumentCaptor.forClass(Properties.class);
+
+ // create the sink, which will create the producer and consumer
+ // suppress "try" warning because of unused sink
+ try (KafkaMirroringSink sink = new KafkaMirroringSink(conf)) {
+ mockedStatic.verify(
+ () -> KafkaCrossDcConf.addSecurityProps(same(conf),
propsCaptor.capture()), times(2));
+ }
+
+ List<Properties> allProps = propsCaptor.getAllValues();
+ Properties producerProps = allProps.get(0);
+ Properties consumerProps = allProps.get(1);
+
+ // check that the captured producer props and the consumer props are
filled in
+ assertTrue(producerProps.containsKey("key.serializer"));
+ assertTrue(consumerProps.containsKey(ConsumerConfig.GROUP_ID_CONFIG));
+
+ for (Properties props : allProps) {
+ assertEquals("SSL",
props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
+ assertEquals(
+ "/tmp/truststore.jks",
props.getProperty(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG));
+ assertEquals("changeit",
props.getProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG));
+ }
+ }
+ }
+}
