Github user srdo commented on a diff in the pull request:

    https://github.com/apache/storm/pull/2538#discussion_r165854506
  
    --- Diff: 
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
 ---
    @@ -396,7 +361,10 @@ private void setWaitingToEmit(ConsumerRecords<K, V> 
consumerRecords) {
                     numPolledRecords);
                 if (kafkaSpoutConfig.getProcessingGuarantee() == 
KafkaSpoutConfig.ProcessingGuarantee.AT_MOST_ONCE) {
                     //Commit polled records immediately to ensure delivery is 
at-most-once.
    -                kafkaConsumer.commitSync();
    +                Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = 
    +                    
createOffsetsToCommitForConsumedOffsets(kafkaConsumer.assignment());
    +                kafkaConsumer.commitSync(offsetsToCommit);
    +                LOG.debug("Committed offsets {} to Kafka", 
offsetsToCommit);
    --- End diff --
    
    I don't mind adding it, but is this information useful to the user?


---

Reply via email to