X-Git-Url: http://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessConsumer.java;h=ba8eb279f649984ad4130197b33845a0fc17a60b;hb=377840107151d9c270f7e3a91a118dce4aa1295f;hp=f0e74d39fd1737d59fdaa22f98815baca4c61df3;hpb=12c1ce703ef76b75d995f94d1689c894dde1406a;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 f0e74d3..ba8eb27 100644 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@ -6,6 +6,7 @@ import org.apache.kafka.clients.consumer.*; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.RecordDeserializationException; import org.apache.kafka.common.errors.WakeupException; +import org.springframework.stereotype.Component; import javax.annotation.PreDestroy; import java.time.Duration; @@ -17,9 +18,10 @@ import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +@Component @Slf4j @RequiredArgsConstructor -public class EndlessConsumer implements Runnable +public class EndlessConsumer implements Runnable { private final ExecutorService executor; private final String id; @@ -71,7 +73,8 @@ public class EndlessConsumer implements Runnable } 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) @@ -85,6 +88,7 @@ public class EndlessConsumer implements Runnable offset, e.getCause().toString()); + consumer.commitSync(); shutdown(e); } catch(Exception e)