fix: Use correct user-defined groupId in Kafka adapter (#3409)

apache · Jan 9, 2025 · c36e260 · c36e260
1 parent 19eeb1a
commit c36e260
Show file tree

Hide file tree

Showing 5 changed files with 12 additions and 42 deletions.
diff --git a/...c/main/java/org/apache/streampipes/extensions/connectors/kafka/adapter/KafkaProtocol.java b/...c/main/java/org/apache/streampipes/extensions/connectors/kafka/adapter/KafkaProtocol.java
@@ -31,7 +31,7 @@
 import org.apache.streampipes.extensions.api.extractor.IAdapterParameterExtractor;
 import org.apache.streampipes.extensions.api.extractor.IStaticPropertyExtractor;
 import org.apache.streampipes.extensions.api.runtime.SupportsRuntimeConfig;
-import org.apache.streampipes.extensions.connectors.kafka.shared.kafka.KafkaAdapterConfig;
+import org.apache.streampipes.extensions.connectors.kafka.shared.kafka.KafkaBaseConfig;
 import org.apache.streampipes.extensions.connectors.kafka.shared.kafka.KafkaConfigExtractor;
 import org.apache.streampipes.extensions.connectors.kafka.shared.kafka.KafkaConfigProvider;
 import org.apache.streampipes.extensions.management.connect.adapter.BrokerEventProcessor;
@@ -74,7 +74,7 @@
 public class KafkaProtocol implements StreamPipesAdapter, SupportsRuntimeConfig {
 
   private static final Logger LOG = LoggerFactory.getLogger(KafkaProtocol.class);
-  private KafkaAdapterConfig config;
+  private KafkaBaseConfig config;
 
   public static final String ID = "org.apache.streampipes.connect.iiot.protocol.stream.kafka";
 
@@ -88,16 +88,14 @@ private void applyConfiguration(IStaticPropertyExtractor extractor) {
     this.config = new KafkaConfigExtractor().extractAdapterConfig(extractor, true);
   }
 
-  private Consumer<byte[], byte[]> createConsumer(KafkaAdapterConfig kafkaConfig) throws KafkaException {
+  private Consumer<byte[], byte[]> createConsumer(KafkaBaseConfig kafkaConfig) throws KafkaException {
     final Properties props = new Properties();
 
     kafkaConfig.getConfigAppenders().forEach(c -> c.appendConfig(props));
 
     props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
         kafkaConfig.getKafkaHost() + ":" + kafkaConfig.getKafkaPort());
 
-    props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConfig.getGroupId());
-
     props.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 6000);
 
     props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,

diff --git a/...a/org/apache/streampipes/extensions/connectors/kafka/shared/kafka/KafkaAdapterConfig.java b/...a/org/apache/streampipes/extensions/connectors/kafka/shared/kafka/KafkaAdapterConfig.java
diff --git a/...org/apache/streampipes/extensions/connectors/kafka/shared/kafka/KafkaConfigExtractor.java b/...org/apache/streampipes/extensions/connectors/kafka/shared/kafka/KafkaConfigExtractor.java
@@ -28,6 +28,7 @@
 import org.apache.streampipes.messaging.kafka.security.KafkaSecuritySaslConfigAppender;
 import org.apache.streampipes.model.staticproperty.StaticPropertyAlternatives;
 
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 
 import java.util.ArrayList;
@@ -51,22 +52,24 @@
 
 public class KafkaConfigExtractor {
 
-  public KafkaAdapterConfig extractAdapterConfig(IStaticPropertyExtractor extractor,
+  public KafkaBaseConfig extractAdapterConfig(IStaticPropertyExtractor extractor,
                                                  boolean containsTopic) {
 
-    var config = extractCommonConfigs(extractor, new KafkaAdapterConfig());
+    var config = extractCommonConfigs(extractor, new KafkaBaseConfig());
 
     var topic = "";
     if (containsTopic) {
       topic = extractor.selectedSingleValue(TOPIC_KEY, String.class);
     }
     config.setTopic(topic);
 
+    var groupId = "";
     if (extractor.selectedAlternativeInternalId(CONSUMER_GROUP).equals(RANDOM_GROUP_ID)) {
-      config.setGroupId("StreamPipesKafkaConsumer" + System.currentTimeMillis());
+      groupId = "StreamPipesKafkaConsumer" + System.currentTimeMillis();
     } else {
-      config.setGroupId(extractor.singleValueParameter(GROUP_ID_INPUT, String.class));
+      groupId = extractor.singleValueParameter(GROUP_ID_INPUT, String.class);
     }
+    config.getConfigAppenders().add(new SimpleConfigAppender(Map.of(ConsumerConfig.GROUP_ID_CONFIG, groupId)));
 
     StaticPropertyAlternatives alternatives = extractor.getStaticPropertyByName(AUTO_OFFSET_RESET_CONFIG,
         StaticPropertyAlternatives.class);

diff --git a/...messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaConsumer.java b/...messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaConsumer.java
@@ -82,6 +82,7 @@ public void run() {
 
     Properties props = makeProperties(protocol, appenders);
 
+    LOG.info("Using kafka properties: {}", props.toString());
     KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props);
     if (!patternTopic) {
       consumer.subscribe(Collections.singletonList(topic));

diff --git a/...ka/src/main/java/org/apache/streampipes/messaging/kafka/config/AbstractConfigFactory.java b/...ka/src/main/java/org/apache/streampipes/messaging/kafka/config/AbstractConfigFactory.java
@@ -48,6 +48,6 @@ public Properties buildProperties(List<KafkaConfigAppender> appenders) {
     Properties props = makeDefaultProperties();
     appenders.forEach(appender -> appender.appendConfig(props));
 
-    return  props;
+    return props;
   }
 }