X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessConsumer.java;h=92802b981f5bcf465758c9b257949ad6b104f8b6;hb=refs%2Ftags%2Fsumup-adder--drop-duplicates---lvm-2-tage;hp=63fc10f98e4808f22f9c27f2cab828ef7f41aee9;hpb=384ed293b9d6d71f68031f92bd7bf60670a7ba4b;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 63fc10f..92802b9 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 CommittingConsumerRebalanceListener rebalanceListener; + private final RebalanceListener 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,11 +67,14 @@ public class EndlessConsumer implements Runnable consumed++; } + + rebalanceListener.beforeNextPoll(); } } catch(WakeupException e) { - log.info("{} - RIIING! Request to stop consumption.", id); + log.info("{} - RIIING! Request to stop consumption - commiting current offsets!", id); + consumer.commitSync(); shutdown(); } catch(RecordDeserializationException e) @@ -86,12 +88,12 @@ public class EndlessConsumer implements Runnable offset, e.getCause().toString()); + consumer.commitSync(); shutdown(e); } 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