Github user reiabreu commented on a diff in the pull request:
https://github.com/apache/storm/pull/2593#discussion_r178510737
--- Diff:
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
---
@@ -484,8 +484,11 @@ private boolean emitOrRetryTuple(ConsumerRecord<K, V>
record) {
return true;
}
} else {
+ /*if a null tuple is not configured to be emitted, it
should be marked as emitted and acked immediately
+ * to allow its offset to be commited to Kafka*/
LOG.debug("Not emitting null tuple for record [{}] as
defined in configuration.", record);
msgId.setEmitted(false);
--- End diff --
I concur. I'll refactor it.
---
