Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1808#discussion_r96907232
--- Diff:
external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java
---
@@ -325,15 +310,19 @@ private boolean
emitTupleIfNotEmitted(ConsumerRecord<K, V> record) {
} else {
boolean isScheduled = retryService.isScheduled(msgId);
if (!isScheduled || retryService.isReady(msgId)) { // not
scheduled <=> never failed (i.e. never emitted) or ready to be retried
- final List<Object> tuple =
tuplesBuilder.buildTuple(record);
- kafkaSpoutStreams.emit(collector, tuple, msgId);
+ final List<Object> tuple =
kafkaSpoutConfig.getTranslator().apply(record);
+ if (tuple instanceof KafkaTuple) {
--- End diff --
Yes, and it is documented in RecordTranslator. I will add in more
documentation on it though.
The reason specifically for this was because the spout is not able to keep
track of a single message being emitted to multiple streams. It would get
confused and ack it before it was truly done. This makes it impossible for
that to happen. What is more the built in record translators should cover 99%
of the use cases, so the fact that it is not super well documented should be
more of a corner case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
