[STORM-2229] KafkaSpout does not resend failed tuples

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Duplicate
• Affects Version/s: 1.0.0, 1.0.1, 1.0.2
• Component/s: storm-kafka-client
• Assignee: Unassigned
• Reporter: Matthias Klein
• Created: 01/Dec/16 07:09
• Updated: 19/Feb/18 19:35
• Resolved: 01/Dec/16 18:44

Description

When the topology fails a tuple, it is never resent by the KafkaSpout. This can easily be shown by constructing a small topology failing every tuple.

Apparent reason:

public class KafkaSpout<K, V> extends BaseRichSpout {
//...
private void doSeekRetriableTopicPartitions() {
        final Set<TopicPartition> retriableTopicPartitions = retryService.retriableTopicPartitions();

        for (TopicPartition rtp : retriableTopicPartitions) {
            final OffsetAndMetadata offsetAndMeta = acked.get(rtp).findNextCommitOffset();
            if (offsetAndMeta != null) {
                kafkaConsumer.seek(rtp, offsetAndMeta.offset() + 1);  // seek to the next offset that is ready to commit in next commit cycle
            } else {
                kafkaConsumer.seekToEnd(toArrayList(rtp));    // Seek to last committed offset <== Does seek to end of partition
            }
        }
    }

The code seeks to the end of the partition instead of seeking to the first uncommited offset.

Preliminary fix (worked for me, but needs to be checked by an expert)

    private void doSeekRetriableTopicPartitions() {
        final Set<TopicPartition> retriableTopicPartitions = retryService.retriableTopicPartitions();

        for (TopicPartition rtp : retriableTopicPartitions) {
            final OffsetAndMetadata offsetAndMeta = acked.get(rtp).findNextCommitOffset();
            if (offsetAndMeta != null) {
                kafkaConsumer.seek(rtp, offsetAndMeta.offset() + 1);  // seek to the next offset that is ready to commit in next commit cycle
            } else {
                OffsetAndMetadata committed = kafkaConsumer.committed(rtp);
                if(committed == null) {
                    // No offsets commited yet for this partition - start from beginning 
                    kafkaConsumer.seekToBeginning(toArrayList(rtp));
                } else {
                   // Seek to first uncommitted offset
                    kafkaConsumer.seek(rtp, committed.offset() + 1);
                }
            }
        }
    }

Issue Links

• duplicates: STORM-2087 Storm-kafka-client: Failed tuples are not always replayed
• is duplicated by: STORM-2077 KafkaSpout doesn't retry failed tuples