Log at error level when the kafka spout gives up on a tuple; document max-retry give-up semantics

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("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
             tupleListener.onMaxRetryReached(msgId);

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).
      * <p/>
      * 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)

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.
      */

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<String, Object> 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();