[hotfix] Modify assignment of groupId
@@ -19,7 +19,6 @@ package org.apache.streampipes.messaging.kafka.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
import java.util.Properties;
@@ -42,7 +41,7 @@ public class ConsumerConfigFactory extends AbstractConfigFactory {
public Properties makeDefaultProperties() {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getBrokerUrl());
- props.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, getConfigOrDefault(protocol::getGroupId, UUID.randomUUID().toString()));
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, ENABLE_AUTO_COMMIT_CONFIG_DEFAULT);
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,
AUTO_COMMIT_INTERVAL_MS_CONFIG_DEFAULT);
