Kafka Backend group ID configuration make easier (#391)

* use julie.instance.id  in case group.id is not defined for the kafka backend

* make default variable as consumer group id for kafka backend consumer, by default named julieops, if not it can be configured via property file or env var

* ammend kafka tests, not to use same group.id, so consumption is done back from beginiing

src/main/java/com/purbon/kafka/topology/Configuration.java

@@ -565,4 +565,8 @@ private void generateRandomJulieInstanceId() {
               .toString();
     }
   }
+
+  public String getKafkaBackendConsumerGroupId() {
+    return config.getString(JULIE_KAFKA_CONSUMER_GROUP_ID);
+  }
 }

src/main/java/com/purbon/kafka/topology/Constants.java

@@ -136,5 +136,6 @@ public class Constants {
   public static final String JULIE_ROLES = "julie.roles";
 
   public static final String JULIE_KAFKA_CONFIG_TOPIC = "julie.kafka.config.topic";
+  public static final String JULIE_KAFKA_CONSUMER_GROUP_ID = "julie.kafka.consumer.group.id";
   public static final String JULIE_INSTANCE_ID = "julie.instance.id";
 }

src/main/java/com/purbon/kafka/topology/backend/kafka/KafkaBackendConsumer.java

@@ -37,9 +37,8 @@ public void configure() {
     var serde = new JsonDeserializer<>(BackendState.class);
     consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, serde.getClass());
     consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-    consumerProperties.put(
-        GROUP_ID_CONFIG,
-        consumerProperties.get(GROUP_ID_CONFIG).toString() + System.currentTimeMillis());
+
+    consumerProperties.put(GROUP_ID_CONFIG, config.getKafkaBackendConsumerGroupId());
     consumer = new KafkaConsumer<>(consumerProperties);
     consumer.subscribe(Collections.singletonList(config.getJulieKafkaConfigTopic()));
   }

src/main/resources/reference.conf

@@ -86,6 +86,8 @@ julie {
     kafka.config.topic = "__julieops_commands"
     kafka.config.topic = ${?JULIE_KAFKA_CONFIG_TOPIC}
     instance.id = ${?JULIE_INSTANCE_ID}
+    kafka.consumer.group.id = "julieops"
+    kafka.consumer.group.id = ${?JULIE_KAFKA_CONSUMER_GROUP_ID}
 }
 
 confluent {

src/test/java/com/purbon/kafka/topology/integration/backend/KafkaBackendIT.java

@@ -2,6 +2,7 @@
 
 import static com.purbon.kafka.topology.CommandLineInterface.BROKERS_OPTION;
 import static com.purbon.kafka.topology.Constants.JULIE_INSTANCE_ID;
+import static com.purbon.kafka.topology.Constants.JULIE_KAFKA_CONSUMER_GROUP_ID;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.purbon.kafka.topology.Configuration;
@@ -39,7 +40,6 @@ public void before() throws IOException {
 
     props = new Properties();
     props.put(JULIE_INSTANCE_ID, "1234");
-    props.put(ConsumerConfig.GROUP_ID_CONFIG, "group.id");
 
     Map<String, Object> saslConfig =
         ContainerTestUtils.getSaslConfig(
@@ -80,6 +80,11 @@ public void testExpectedFlow() throws IOException, InterruptedException {
     backend.close();
 
     KafkaBackend newBackend = new KafkaBackend();
+
+    HashMap<String, String> cliOps = new HashMap<>();
+    cliOps.put(BROKERS_OPTION, container.getBootstrapServers());
+    props.put(JULIE_KAFKA_CONSUMER_GROUP_ID, "julieops"+System.currentTimeMillis());
+    Configuration config = new Configuration(cliOps, props);
     newBackend.configure(config);
 
     Thread.sleep(3000);