Github user hmcl commented on a diff in the pull request: https://github.com/apache/storm/pull/2380#discussion_r147015407 --- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java --- @@ -438,55 +440,53 @@ private void commitOffsetsForAckedTuples() { // ======== Ack ======= @Override public void ack(Object messageId) { - if (!isAtLeastOnce()) { - // Only need to keep track of acked tuples if commits are done based on acks - return; - } - + // Only need to keep track of acked tuples if commits to Kafka are done after a tuple ack is received final KafkaSpoutMessageId msgId = (KafkaSpoutMessageId) messageId; - if (!emitted.contains(msgId)) { - if (msgId.isEmitted()) { - LOG.debug("Received ack for message [{}], associated with tuple emitted for a ConsumerRecord that " - + "came from a topic-partition that this consumer group instance is no longer tracking " - + "due to rebalance/partition reassignment. No action taken.", msgId); + if (isAtLeastOnceProcessing()) { + if (!emitted.contains(msgId)) { + if (msgId.isEmitted()) { + LOG.debug("Received ack for message [{}], associated with tuple emitted for a ConsumerRecord that " + + "came from a topic-partition that this consumer group instance is no longer tracking " + + "due to rebalance/partition reassignment. No action taken.", msgId); + } else { + LOG.debug("Received direct ack for message [{}], associated with null tuple", msgId); + } } else { - LOG.debug("Received direct ack for message [{}], associated with null tuple", msgId); + Validate.isTrue(!retryService.isScheduled(msgId), "The message id " + msgId + " is queued for retry while being acked." + + " This should never occur barring errors in the RetryService implementation or the spout code."); + offsetManagers.get(msgId.getTopicPartition()).addToAckMsgs(msgId); + emitted.remove(msgId); } - } else { - Validate.isTrue(!retryService.isScheduled(msgId), "The message id " + msgId + " is queued for retry while being acked." - + " This should never occur barring errors in the RetryService implementation or the spout code."); - offsetManagers.get(msgId.getTopicPartition()).addToAckMsgs(msgId); - emitted.remove(msgId); + tupleListener.onAck(msgId); } - tupleListener.onAck(msgId); } // ======== Fail ======= @Override public void fail(Object messageId) { - if (!isAtLeastOnce()) { - // Only need to keep track of failed tuples if commits are done based on acks - return; - } + // Only need to keep track of failed tuples if commits to Kafka are done after a tuple ack is received + if (isAtLeastOnceProcessing()) { --- End diff -- Same comment above
---