X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessConsumer.java;h=3ff479c94aaa019342dd7fb37a1029db786b92d3;hb=627763878b235ba168c7f55a1ef448851b027bfc;hp=02385211cdf8470df6973f6bb608095dcebe501f;hpb=1c6d263c619010d23bf502c14dda45db11a2baf6;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 0238521..3ff479c 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 RebalanceListener rebalanceListener; private final RecordHandler recordHandler; private final Lock lock = new ReentrantLock(); @@ -42,6 +42,7 @@ public class EndlessConsumer implements Runnable try { log.info("{} - Subscribing to topic {}", id, topic); + rebalanceListener.enableCommits(); consumer.subscribe(Arrays.asList(topic), rebalanceListener); while (true) @@ -67,14 +68,11 @@ public class EndlessConsumer implements Runnable consumed++; } - - rebalanceListener.beforeNextPoll(); } } catch(WakeupException e) { - log.info("{} - RIIING! Request to stop consumption - commiting current offsets!", id); - consumer.commitSync(); + log.info("{} - RIIING! Request to stop consumption.", id); shutdown(); } catch(RecordDeserializationException e) @@ -88,12 +86,12 @@ public class EndlessConsumer implements Runnable offset, e.getCause().toString()); - consumer.commitSync(); shutdown(e); } catch(Exception e) { - log.error("{} - Unexpected error: {}", id, e.toString(), e); + log.error("{} - Unexpected error: {}, disabling commits", id, e.toString(), e); + rebalanceListener.disableCommits(); shutdown(e); } finally