X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationConfiguration.java;h=9b06b09367f7cf748dd6d770bd517ee4fcbede38;hb=ecd63311e0b6af698185bcf1e085f2b3237bd264;hp=54e9b893f45ffca7303b58bf63b8d69348f65fca;hpb=e53004f3133b737699e995a3b18fff28203a0e8c;p=demos%2Fkafka%2Ftraining

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index 54e9b89..9b06b09 100644
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -8,6 +8,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import java.time.Clock;
 import java.util.Properties;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -41,6 +42,8 @@ public class ApplicationConfiguration
             repository,
             properties.getClientId(),
             properties.getTopic(),
+            Clock.systemDefaultZone(),
+            properties.getCommitInterval(),
             kafkaConsumer,
             handler);
   }
@@ -57,6 +60,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("group.id", properties.getGroupId());
     props.put("client.id", properties.getClientId());
     props.put("enable.auto.commit", false);