Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2593#discussion_r178562679
--- Diff:
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
---
@@ -570,20 +572,25 @@ public void ack(Object messageId) {
// Only need to keep track of acked tuples if commits to Kafka are
controlled by
// tuple acks, which happens only for at-least-once processing
semantics
final KafkaSpoutMessageId msgId = (KafkaSpoutMessageId) messageId;
- if (!emitted.contains(msgId)) {
- if (msgId.isEmitted()) {
+ if (!msgId.isNullTuple()) {
--- End diff --
Great, that sounds good.
---
