Merge pull request #36 from ssaip/develop

add group.id.prefix in consumer.yaml #35
patschuh · Feb 6, 2023 · a09d061 · a09d061
2 parents 5874745 + 4c69b32
commit a09d061
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/src/main/java/at/esque/kafka/handlers/ConsumerHandler.java b/src/main/java/at/esque/kafka/handlers/ConsumerHandler.java
@@ -55,8 +55,10 @@ public void setRegisteredConsumers(Map<UUID, KafkaConsumer> registeredConsumers)
     public UUID registerConsumer(ClusterConfig config, TopicMessageTypeConfig topicMessageTypeConfig, Map<String, String> consumerConfigs) throws MissingSchemaRegistryException {
         Properties consumerProps = new Properties();
         consumerProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootStrapServers());
+        String groupIdPrefix = Optional.ofNullable(consumerConfigs.get("group.id.prefix"))
+                .orElse("kafkaesque-");
         UUID consumerId = UUID.randomUUID();
-        consumerProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "kafkaesque-" + consumerId);
+        consumerProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupIdPrefix + consumerId);
         consumerProps.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
         consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KafkaEsqueDeserializer.class);
         consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaEsqueDeserializer.class);