Vorlage vereinfacht: Rebalance-Listener entfernt

[demos/kafka/training] / src / main / java / de / juplo / kafka / ApplicationConfiguration.java

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index 1c69760..e887987 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -33,25 +33,10 @@ public class ApplicationConfiguration
     return new AdderResults();
   }
 
-  @Bean
-  public ApplicationRebalanceListener rebalanceListener(
-      ApplicationRecordHandler recordHandler,
-      AdderResults adderResults,
-      StateRepository stateRepository,
-      ApplicationProperties properties)
-  {
-    return new ApplicationRebalanceListener(
-        recordHandler,
-        adderResults,
-        stateRepository,
-        properties.getClientId());
-  }
-
   @Bean
   public EndlessConsumer<String, String> endlessConsumer(
       KafkaConsumer<String, String> kafkaConsumer,
       ExecutorService executor,
-      ApplicationRebalanceListener rebalanceListener,
       ApplicationRecordHandler recordHandler,
       ApplicationProperties properties)
   {
@@ -61,7 +46,6 @@ public class ApplicationConfiguration
             properties.getClientId(),
             properties.getTopic(),
             kafkaConsumer,
-            rebalanceListener,
             recordHandler);
   }
 
@@ -77,7 +61,7 @@ public class ApplicationConfiguration
     Properties props = new Properties();
 
     props.put("bootstrap.servers", properties.getBootstrapServer());
-    props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
+    props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.StickyAssignor");
     props.put("group.id", properties.getGroupId());
     props.put("client.id", properties.getClientId());
     props.put("auto.offset.reset", properties.getAutoOffsetReset());