This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new 27ae976bf Pass only the required Kafka consumer properties to the
spout lag monitor
27ae976bf is described below
commit 27ae976bf962ee8608e3e66e3821e3a52951b79d
Author: Richard Zowalla <[email protected]>
AuthorDate: Fri Aug 21 14:21:44 2026 +0200
Pass only the required Kafka consumer properties to the spout lag monitor
---
.../org/apache/storm/utils/TopologySpoutLag.java | 39 +++++++++-
.../apache/storm/utils/TopologySpoutLagTest.java | 89 ++++++++++++++++++++++
2 files changed, 126 insertions(+), 2 deletions(-)
diff --git a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
index d6bc36052..d9bbc92b7 100644
--- a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
+++ b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
@@ -16,6 +16,7 @@
package org.apache.storm.utils;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
import java.io.File;
@@ -52,6 +53,28 @@ public class TopologySpoutLag {
private static final String SECURITY_PROTOCOL_CONFIG = CONFIG_KEY_PREFIX +
"security.protocol";
private static final Set<String> ALL_CONFIGS = new
HashSet<>(Arrays.asList(TOPICS_CONFIG, GROUPID_CONFIG,
BOOTSTRAP_CONFIG, SECURITY_PROTOCOL_CONFIG));
+ // The spout json_conf comes from the submitted topology, while
storm-kafka-monitor runs on the UI host.
+ // Only the connection settings the monitor actually needs to reach the
brokers are forwarded to it; anything
+ // else (deserializers, interceptor.classes, metric.reporters,
sasl.jaas.config, callback handler classes, ...)
+ // is dropped, so the monitor keeps using its own defaults rather than
classes named by the topology.
+ private static final Set<String> ALLOWED_EXTRA_PROPERTIES = new
HashSet<>(Arrays.asList(
+ "client.id",
+ "request.timeout.ms",
+ "session.timeout.ms",
+ "sasl.mechanism",
+ "sasl.kerberos.service.name",
+ "ssl.protocol",
+ "ssl.provider",
+ "ssl.enabled.protocols",
+ "ssl.cipher.suites",
+ "ssl.endpoint.identification.algorithm",
+ "ssl.truststore.type",
+ "ssl.truststore.location",
+ "ssl.truststore.password",
+ "ssl.keystore.type",
+ "ssl.keystore.location",
+ "ssl.keystore.password",
+ "ssl.key.password"));
private static final Logger LOGGER =
LoggerFactory.getLogger(TopologySpoutLag.class);
// The storm-kafka-monitor jars ship only in the full binary distribution;
users of the lite
@@ -110,14 +133,26 @@ public class TopologySpoutLag {
return commands;
}
- private static File createExtraPropertiesFile(Map<String, Object>
jsonConf) {
+ @VisibleForTesting
+ static File createExtraPropertiesFile(Map<String, Object> jsonConf) {
File file = null;
Map<String, String> extraProperties = new HashMap<>();
+ List<String> droppedProperties = new ArrayList<>();
for (Map.Entry<String, Object> conf : jsonConf.entrySet()) {
if (conf.getKey().startsWith(CONFIG_KEY_PREFIX) &&
!ALL_CONFIGS.contains(conf.getKey())) {
-
extraProperties.put(conf.getKey().substring(CONFIG_KEY_PREFIX.length()),
conf.getValue().toString());
+ String consumerKey =
conf.getKey().substring(CONFIG_KEY_PREFIX.length());
+ if (ALLOWED_EXTRA_PROPERTIES.contains(consumerKey)) {
+ extraProperties.put(consumerKey,
conf.getValue().toString());
+ } else {
+ droppedProperties.add(consumerKey);
+ }
}
}
+ // The UI polls the lag endpoint, so log the dropped keys once per
call rather than one line each.
+ if (!droppedProperties.isEmpty()) {
+ LOGGER.info("Not passing consumer properties {} from the topology
to the Kafka spout lag monitor, "
+ + "only these properties are passed on: {}",
droppedProperties, ALLOWED_EXTRA_PROPERTIES);
+ }
if (!extraProperties.isEmpty()) {
try {
file = Files.createTempFile("kafka-consumer-extra",
"props").toFile();
diff --git
a/storm-core/test/jvm/org/apache/storm/utils/TopologySpoutLagTest.java
b/storm-core/test/jvm/org/apache/storm/utils/TopologySpoutLagTest.java
new file mode 100644
index 000000000..864596877
--- /dev/null
+++ b/storm-core/test/jvm/org/apache/storm/utils/TopologySpoutLagTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.storm.utils;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+
+public class TopologySpoutLagTest {
+
+ private Properties loadProperties(File file) throws IOException {
+ Properties properties = new Properties();
+ try (InputStream in = new FileInputStream(file)) {
+ properties.load(in);
+ }
+ return properties;
+ }
+
+ @Test
+ public void testOnlyKnownConsumerPropertiesArePassedToTheMonitor() throws
Exception {
+ Map<String, Object> jsonConf = new HashMap<>();
+ // Passed to the monitor as command line options instead of via the
properties file.
+ jsonConf.put("config.topics", "topic");
+ jsonConf.put("config.groupid", "group");
+ jsonConf.put("config.bootstrap.servers", "broker:9092");
+ jsonConf.put("config.security.protocol", "SASL_SSL");
+ // Connection settings the monitor needs.
+ jsonConf.put("config.sasl.mechanism", "SCRAM-SHA-512");
+ jsonConf.put("config.ssl.truststore.location",
"/etc/storm/truststore.jks");
+ jsonConf.put("config.client.id", "lag-monitor");
+ // Properties that would make the monitor load classes named by the
topology.
+ jsonConf.put("config.key.deserializer", "org.example.MyDeserializer");
+ jsonConf.put("config.value.deserializer",
"org.example.MyDeserializer");
+ jsonConf.put("config.interceptor.classes",
"org.example.MyInterceptor");
+ jsonConf.put("config.metric.reporters", "org.example.MyReporter");
+ jsonConf.put("config.sasl.jaas.config", "org.example.MyLoginModule
required;");
+ jsonConf.put("config.sasl.login.callback.handler.class",
"org.example.MyCallbackHandler");
+ // Keys without the config. prefix are not consumer properties at all.
+ jsonConf.put("topology.name", "test");
+
+ File file = TopologySpoutLag.createExtraPropertiesFile(jsonConf);
+ try {
+ Properties properties = loadProperties(file);
+ Properties expected = new Properties();
+ expected.put("sasl.mechanism", "SCRAM-SHA-512");
+ expected.put("ssl.truststore.location",
"/etc/storm/truststore.jks");
+ expected.put("client.id", "lag-monitor");
+ assertEquals(expected, properties);
+ } finally {
+ file.delete();
+ }
+ }
+
+ @Test
+ public void testNoPropertiesFileWhenNoKnownConsumerPropertiesAreSet() {
+ Map<String, Object> jsonConf = new HashMap<>();
+ jsonConf.put("config.topics", "topic");
+ jsonConf.put("config.groupid", "group");
+ jsonConf.put("config.bootstrap.servers", "broker:9092");
+ jsonConf.put("config.interceptor.classes",
"org.example.MyInterceptor");
+
+ assertNull(TopologySpoutLag.createExtraPropertiesFile(jsonConf));
+ }
+}
