From cf612055575f3aec3fc743b4a8de97f57c691088 Mon Sep 17 00:00:00 2001 From: L1nq0 Date: Wed, 16 Sep 2026 09:44:58 +0800 Subject: [PATCH 1/2] Log at error level when the kafka spout gives up on a tuple; document max-retry give-up semantics --- .../main/java/org/apache/storm/kafka/spout/KafkaSpout.java | 3 ++- .../kafka/spout/KafkaSpoutRetryExponentialBackoff.java | 6 +++++- .../org/apache/storm/kafka/spout/KafkaTupleListener.java | 4 +++- 3 files changed, 10 insertions(+), 3 deletions(-) diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java index 73582dd4c8a..cb4fcbcf1ba 100644 --- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java +++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java @@ -605,7 +605,8 @@ public void fail(Object messageId) { msgId.incrementNumFails(); if (!retryService.schedule(msgId)) { - LOG.debug("Reached maximum number of retries. Message [{}] being marked as acked.", msgId); + LOG.error("Reached maximum number of retries. Giving up on message [{}]: the tuple will be acked and its offset and later " + + "offsets may be committed even though the record was not processed.", msgId); // this tuple should be removed from emitted only inside the ack() method. This is to ensure // that the OffsetManager for that TopicPartition is updated and allows commit progression tupleListener.onMaxRetryReached(msgId); diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java index 8a2f54303e1..76b7e392103 100644 --- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java +++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java @@ -156,7 +156,11 @@ public String toString() { * where failCount = 1, 2, 3, ... nextRetry = Min(nextRetry, currentTime + maxDelay). *

* By specifying a value for maxRetries lower than Integer.MAX_VALUE, the user decides to sacrifice guarantee of delivery for the - * previous polled records in favor of processing more records. + * previous polled records in favor of processing more records. Setting a finite limit also stops the spout from retrying forever + * tuples that fail every time they are emitted, e.g. tuples the receiving worker drops because they cannot be deserialized: + * once the limit is reached, the tuple is acked and offsets past it can be committed. These tuples are reported to + * {@link KafkaTupleListener#onMaxRetryReached(KafkaSpoutMessageId)}, which receives the topic, partition and offset needed + * to fetch the record from Kafka again. * * @param initialDelay initial delay of the first retry * @param delayPeriod the time interval that is the ratio of the exponential backoff formula (geometric progression) diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaTupleListener.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaTupleListener.java index 10014831285..e9e5cb7a7a3 100644 --- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaTupleListener.java +++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaTupleListener.java @@ -74,7 +74,9 @@ public interface KafkaTupleListener extends Serializable { void onRetry(KafkaSpoutMessageId msgId); /** - * Called when the maximum number of retries have been reached. + * Called when the maximum number of retries have been reached. The tuple is acked right after this callback and + * commits can then move past its offset; this is the last point at which the record can be retained. The msgId + * identifies the record by topic, partition and offset. * * @param msgId The id of the tuple in the spout. */ From 971c768c4190e183981ef49f57d48c7d6c751236 Mon Sep 17 00:00:00 2001 From: L1nq0 Date: Wed, 16 Sep 2026 16:34:16 +0800 Subject: [PATCH 2/2] Reword the give-up error to hold for any retry service and pin the max-retry ack ordering KafkaSpoutRetryService#schedule may decline a message for reasons other than reaching the retry limit, so the give-up log no longer names the reason. KafkaSpoutRetryLimitTest now verifies that onMaxRetryReached runs before the tuple is acked, matching the documented contract. --- .../java/org/apache/storm/kafka/spout/KafkaSpout.java | 2 +- .../storm/kafka/spout/KafkaSpoutRetryLimitTest.java | 9 +++++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java index cb4fcbcf1ba..b4dca55af22 100644 --- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java +++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java @@ -605,7 +605,7 @@ public void fail(Object messageId) { msgId.incrementNumFails(); if (!retryService.schedule(msgId)) { - LOG.error("Reached maximum number of retries. Giving up on message [{}]: the tuple will be acked and its offset and later " + LOG.error("The retry service will not retry message [{}]: the tuple will be acked and its offset and later " + "offsets may be committed even though the record was not processed.", msgId); // this tuple should be removed from emitted only inside the ack() method. This is to ensure // that the OffsetManager for that TopicPartition is updated and allows commit progression diff --git a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRetryLimitTest.java b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRetryLimitTest.java index 381796b7340..6d4a9e8033c 100644 --- a/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRetryLimitTest.java +++ b/external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutRetryLimitTest.java @@ -54,6 +54,7 @@ public class KafkaSpoutRetryLimitTest { private final long offsetCommitPeriodMs = 2_000; private final TopologyContext contextMock = mock(TopologyContext.class); private final SpoutOutputCollector collectorMock = mock(SpoutOutputCollector.class); + private final KafkaTupleListener tupleListener = mock(KafkaTupleListener.class); private final Map conf = new HashMap<>(); private final TopicPartition partition = new TopicPartition(SingleTopicKafkaSpoutConfiguration.TOPIC, 1); @Mock @@ -72,6 +73,7 @@ public void setUp() { spoutConfig = createKafkaSpoutConfigBuilder(mock(TopicFilter.class), mock(ManualPartitioner.class), -1) .setOffsetCommitPeriodMs(offsetCommitPeriodMs) .setRetry(ZERO_RETRIES_RETRY_SERVICE) + .setTupleListener(tupleListener) .build(); } @@ -99,6 +101,13 @@ public void testFailingTupleCompletesAckAfterRetryLimitIsMet() { spout.fail(messageId); } + // The give-up contract: onMaxRetryReached runs before the tuple is acked. + InOrder giveUpOrder = inOrder(tupleListener); + for (KafkaSpoutMessageId messageId : messageIds.getAllValues()) { + giveUpOrder.verify(tupleListener).onMaxRetryReached(messageId); + giveUpOrder.verify(tupleListener).onAck(messageId); + } + // Advance time and then trigger call to kafka consumer commit Time.advanceTime(KafkaSpout.TIMER_DELAY_MS + offsetCommitPeriodMs); spout.nextTuple();