X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessConsumer.java;h=f0e74d39fd1737d59fdaa22f98815baca4c61df3;hb=5805651c16e07a0710b88c2822894941f67c313e;hp=58374f49a9634704e2e477e6c99b323149dc2dad;hpb=a2e8fc924e5b472d6b90c42d311514f91ea452f1;p=demos%2Fkafka%2Ftraining diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java index 58374f4..f0e74d3 100644 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@ -25,7 +25,7 @@ public class EndlessConsumer implements Runnable private final String id; private final String topic; private final Consumer consumer; - private final PollIntervalAwareConsumerRebalanceListener rebalanceListener; + private final ConsumerRebalanceListener rebalanceListener; private final RecordHandler recordHandler; private final Lock lock = new ReentrantLock(); @@ -42,7 +42,6 @@ public class EndlessConsumer implements Runnable try { log.info("{} - Subscribing to topic {}", id, topic); - rebalanceListener.enableCommits(); consumer.subscribe(Arrays.asList(topic), rebalanceListener); while (true) @@ -68,13 +67,11 @@ public class EndlessConsumer implements Runnable consumed++; } - - rebalanceListener.beforeNextPoll(); } } catch(WakeupException e) { - log.info("{} - RIIING! Request to stop consumption - commiting current offsets!", id); + log.info("{} - RIIING! Request to stop consumption.", id); shutdown(); } catch(RecordDeserializationException e) @@ -92,8 +89,7 @@ public class EndlessConsumer implements Runnable } catch(Exception e) { - log.error("{} - Unexpected error: {}, disabling commits", id, e.toString(), e); - rebalanceListener.disableCommits(); + log.error("{} - Unexpected error: {}", id, e.toString(), e); shutdown(e); } finally