LEGLINK-186: Stop stamping a retry due-time on dead-lettered records

The backoff timestamp is a due-time the -Retry listener waits on, so it is
a promise that the record will be delivered again. It was stamped before
the routing decision was known, so every dead letter carried one - a
delivery that was never scheduled. Poison landed on -Error advertising a
retry ~20s in the future that nothing would ever perform.

Stamping is now split by what each header means:

  attempts           history - how many times the record ran. Stamped
                     unconditionally, survives onto the dead letter.
  backoff-timestamp  a promise - stamped only when the decision is RETRY.

The remove() before the conditional is load-bearing. An exhausted record
arrives carrying the due-time from its previous retry hop, so declining to
stamp a new one is not enough; the stale one has to be cleared or it rides
onto the DLT looking current.

No routing or timing change: the due-time is computed identically for the
RETRY case, and the other three decisions never used it.

Nothing reads these headers off -Error today - the DLT listener was
suppressed earlier in this branch - which is why this was invisible. It
matters for the dead-letter replay capability we want: replay tooling
reading -Error would otherwise find attempts=1 and a pending due-time on
records that are finished.

Renamed RetryTopicRecovererLoggingTest -> RetryTopicRecovererDecisionTest;
it now covers the topic, the log line and the headers, i.e. everything
derived from the Decision. Five header tests added, three of which failed
before this change. The two that already passed guard the other direction:
a retried record must still carry a future due-time, and a dead letter must
still record its attempt count.

Tests: shared 12, measureeval 130, validation 153, 0 failures.

Verified live on a rebuilt image, EvaluationRequested ladder:
  18:26:54.091  Retry attempt 1/4  [EvaluationRequested]
  18:27:14.095  Retry attempt 2/4  [EvaluationRequested-Retry]   +20.004s
  18:28:14.098  Retry attempt 3/4  [EvaluationRequested-Retry]   +60.003s
  18:30:14.101  Max retry attempts (4) reached. Routing to DLT   +120.003s
Every interval within 4ms of PT20S/PT60S/PT120S, matching the runs from
before this change. That final hop is the case this fix targets: the record
had genuinely retried three times, so it arrived carrying the due-time its
previous hop stamped, and that stale value had to be cleared rather than
merely not re-stamped.

Claude-Session: https://claude.ai/code/session_01AZubeEYCKfAmDCcgVTr18G

Author: arianamihailescu

…figs/RetryTopicRecovererLoggingTest.java …igs/RetryTopicRecovererDecisionTest.javaJava/measureeval/src/test/java/com/lantanagroup/link/measureeval/configs/RetryTopicRecovererLoggingTest.java renamed to Java/measureeval/src/test/java/com/lantanagroup/link/measureeval/configs/RetryTopicRecovererDecisionTest.java Java/measureeval/src/test/java/com/lantanagroup/link/measureeval/configs/RetryTopicRecovererLoggingTest.java renamed to Java/measureeval/src/test/java/com/lantanagroup/link/measureeval/configs/RetryTopicRecovererDecisionTest.java

@@ -12,6 +12,7 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.header.Header;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -20,6 +21,7 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.retrytopic.RetryTopicHeaders;
 
+import java.math.BigInteger;
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.Collections;
@@ -32,18 +34,18 @@
 import static org.mockito.Mockito.*;
 
 /**
- * The recoverer's log line must describe what actually happened to the record.
+ * Everything the recoverer derives from its routing decision — the topic, the log line, and the retry
+ * headers — must agree with each other.
  *
- * <p>{@link RetryTopicRecoverer} emits the log, but the destination is chosen by the resolver built in
- * {@link RetryTopicRecovererFactory}, which considers three reasons to dead-letter: retries disabled,
- * poison exception, attempts exhausted. A log derived from only one of those three reports a retry that
- * never happens for the other two — the record is already on {@code -Error} while the log says
- * "Retry attempt 1/N", and no dead-letter line is ever emitted.</p>
+ * <p>{@link RetryTopicRecoverer} emits the log and stamps the headers, but the destination is chosen by
+ * the resolver built in {@link RetryTopicRecovererFactory}, which considers three reasons to
+ * dead-letter: retries disabled, poison exception, attempts exhausted. Anything derived from only one of
+ * those three describes a retry that never happens for the other two.</p>
  *
- * <p>These tests drive the real production wiring and assert the log against the topic the record was
- * actually published to, so the two can never drift apart again.</p>
+ * <p>These tests drive the real production wiring and assert the log and the published headers against
+ * the topic the record actually went to, so the three can never drift apart again.</p>
  */
-class RetryTopicRecovererLoggingTest {
+class RetryTopicRecovererDecisionTest {
 
     private static final String RETRY_TOPIC = "ResourcesNormalized-Retry";
     private static final String ERROR_TOPIC = "ResourcesNormalized-Error";
@@ -69,11 +71,19 @@ void detachAppender() {
         recovererLogger.setLevel(originalLevel);
     }
 
-    /** What the recoverer did with a record: where it went, and what it said about it. */
-    private record Outcome(String topic, ILoggingEvent log) {
+    /** What the recoverer did with a record: what it published, and what it said about it. */
+    private record Outcome(ProducerRecord<Object, Object> published, ILoggingEvent log) {
+        String topic() {
+            return published.topic();
+        }
+
         String message() {
             return log.getFormattedMessage();
         }
+
+        Header header(String name) {
+            return published.headers().lastHeader(name);
+        }
     }
 
     private static ConsumerRecord<String, String> record() {
@@ -116,7 +126,7 @@ private Outcome run(KafkaRetryConfig config, ConsumerRecord<String, String> reco
 
         assertEquals(1, appender.list.size(),
                 "recoverer must emit exactly one log line per failure, got: " + appender.list);
-        return new Outcome(captor.getValue().topic(), appender.list.get(0));
+        return new Outcome(captor.getValue(), appender.list.get(0));
     }
 
     @Test
@@ -163,4 +173,69 @@ void transientFailure_withAttemptsRemaining_isLoggedAsARetry() {
         assertTrue(outcome.message().contains("Retry attempt"),
                 "record was retried but not logged as a retry: " + outcome.message());
     }
+
+    // ---- Backoff-timestamp header. It is a due-time: "do not deliver before this instant", read by the
+    //      backoff-aware -Retry listener. A record going to -Error will never be delivered again, so a
+    //      due-time on it describes a retry that was never scheduled. Nothing reads it there today, which
+    //      is exactly why it is worth pinning: dead-letter replay tooling would find headers implying a
+    //      pending retry on records that are finished.
+
+    @Test
+    void retriedRecord_carriesBackoffDueTime() {
+        Outcome outcome = run(retryConfig(3, false), record(), new RuntimeException("transient boom"));
+
+        assertEquals(RETRY_TOPIC, outcome.topic(), "precondition: transient failure must route to -Retry");
+        Header backoff = outcome.header(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP);
+        assertNotNull(backoff, "a record that will be retried must carry the due-time the listener waits on");
+        assertTrue(new BigInteger(backoff.value()).longValue() >= System.currentTimeMillis() - 1_000L,
+                "due-time must be in the future, not a stale instant");
+    }
+
+    @Test
+    void poisonRecord_carriesNoBackoffDueTime() {
+        Outcome outcome = run(retryConfig(3, false), record(), new ValidationException("Query Type is null."));
+
+        assertEquals(ERROR_TOPIC, outcome.topic(), "precondition: poison must route to -Error");
+        assertNull(outcome.header(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP),
+                "a dead-lettered record must not advertise a due-time for a retry that will never happen");
+    }
+
+    @Test
+    void exhaustedRecord_carriesNoBackoffDueTime() {
+        ConsumerRecord<String, String> record = record();
+        stampAttempts(record, 2);
+
+        Outcome outcome = run(retryConfig(3, false), record, new RuntimeException("still failing"));
+
+        assertEquals(ERROR_TOPIC, outcome.topic(), "precondition: exhausted attempts must route to -Error");
+        assertNull(outcome.header(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP),
+                "a dead-lettered record must not advertise a due-time for a retry that will never happen");
+    }
+
+    @Test
+    void exhaustedRecord_dropsTheDueTimeLeftByItsPreviousRetry() {
+        // The realistic path: this record retried before, so it arrives carrying the due-time stamped on
+        // the last hop. Not stamping a new one is not enough — the stale one must be cleared.
+        ConsumerRecord<String, String> record = record();
+        stampAttempts(record, 2);
+        record.headers().add(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP,
+                BigInteger.valueOf(System.currentTimeMillis() + 60_000L).toByteArray());
+
+        Outcome outcome = run(retryConfig(3, false), record, new RuntimeException("still failing"));
+
+        assertEquals(ERROR_TOPIC, outcome.topic(), "precondition: exhausted attempts must route to -Error");
+        assertNull(outcome.header(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP),
+                "the due-time from the previous retry must be removed, not carried onto the dead letter");
+    }
+
+    @Test
+    void deadLetteredRecord_stillRecordsHowManyAttemptsItGot() {
+        // attempts is history, not a promise — it stays, so a dead letter says how many times it ran.
+        Outcome outcome = run(retryConfig(3, false), record(), new ValidationException("Query Type is null."));
+
+        Header attempts = outcome.header(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS);
+        assertNotNull(attempts, "attempts must survive onto the dead letter");
+        assertEquals(1, ByteBuffer.wrap(attempts.value()).getInt(),
+                "poison failed once and was never retried, so it must record exactly one attempt");
+    }
 }

Java/shared/src/main/java/com/lantanagroup/link/shared/kafka/RetryTopicRecoverer.java

@@ -71,21 +71,32 @@ public RetryTopicRecoverer(int maxAttempts, IntToLongFunction backoffMsForAttemp
     public void accept(ConsumerRecord<?, ?> record, Exception exception) {
         Headers headers = record.headers();
 
+        // attempts is history: it records how many times the record ran, so it is stamped
+        // unconditionally and survives onto a dead letter.
         int attempt = currentAttempts(headers) + 1;
         headers.remove(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS);
         headers.add(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS, ByteBuffer.allocate(4).putInt(attempt).array());
 
-        long newBackoff = System.currentTimeMillis() + backoffMsForAttempt.applyAsLong(attempt);
-        headers.remove(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP);
-        headers.add(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP, BigInteger.valueOf(newBackoff).toByteArray());
         if (headers.lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP) == null) {
             headers.add(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP, BigInteger.valueOf(record.timestamp()).toByteArray());
         }
 
         // Exception headers are stamped by the DeadLetterPublishingRecoverer delegate on publish.
         // Evaluated here, after the attempts header is stamped, so this sees the same count the
         // delegate's destination resolver will read a moment later.
-        switch (decide.apply(record, exception)) {
+        Decision decision = decide.apply(record, exception);
+
+        // The backoff timestamp is a due-time the -Retry listener waits on, so it is a promise that the
+        // record will be delivered again. Stamp it only when that is true. A dead letter is never
+        // redelivered, so a due-time on it describes a retry that was never scheduled — clear the one a
+        // previous hop left rather than carrying it onto the DLT.
+        headers.remove(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP);
+        if (decision == Decision.RETRY) {
+            long dueAt = System.currentTimeMillis() + backoffMsForAttempt.applyAsLong(attempt);
+            headers.add(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP, BigInteger.valueOf(dueAt).toByteArray());
+        }
+
+        switch (decision) {
             case RETRY -> logger.info("Retry attempt {}/{} for topic [{}].",
                     attempt, maxAttempts, record.topic());
             case DLT_EXHAUSTED -> logger.warn("Max retry attempts ({}) reached for topic [{}]. Routing to DLT.",