Skip to content

Commit 62857e1

Browse files
arianamihailescuarianamihailescu
committed
LEGLINK-186: Fix retry log/routing divergence and dead-letter listener
Three defects found while exercising the Kafka retry path against a local stack. Routing behaviour is unchanged throughout; all three are about what gets reported and what gets consumed. 1. The recoverer logged a retry for records it was dead-lettering. RetryTopicRecoverer chose its log from `attempt >= maxAttempts` while RetryTopicRecovererFactory chose the destination from three conditions: retries disabled, poison exception, or attempts exhausted. The log knew about only the third, so a poison record was published to -Error while the log read "Retry attempt 1/4" - and no dead-letter line was emitted at all. Verified in a running service: a ValidationException on ResourcesNormalized landed on -Error and reported a retry. Both now derive from one Decision (RETRY / DLT_DISABLED / DLT_POISON / DLT_EXHAUSTED) computed once in the factory and injected into the recoverer, so the log cannot contradict the routing. The enum, rather than a boolean, is what lets the log name which of the three applied. The same defect hid disable-retry-consumer entirely: with retries off, every record dead-letters while every line claims "Retry attempt 1/4". 2. Both services consumed their own -Error topic. Spring provisions a listener for every hop in the retry chain, including the DLT. Nothing consumes dead letters here - there is no @DltHandler - so that container re-read each one, failed again on the same bytes for deserialization poison (a second stack trace per record), and committed the offset, advancing the group past records dead-letter replay needs. Suppressed with autoStartDltHandler(false). Note doNotConfigureDlt() is NOT equivalent and would be a regression: the DLT must stay in the chain because it is the routing target for container-thread poison, which Spring's own error handler resolves via the chain terminal. 3. Validation could not dead-letter a malformed message at all. Its key and value DelegatingByTypeSerializers had no byte[] delegate. A record that fails deserialization reaches the publisher as the raw bytes that could not be parsed, so publication threw SerializationException, DefaultErrorHandler could not recover, and the container redelivered forever - the partition blocked, measured at 872 failed publications per minute, stranding every message behind it. Pre-existing and never covered by a test; measureeval already had both delegates, which is why only validation was affected. Verified live on rebuilt images, both services: retry ladder intact at 20.003s / 60.006s / 120.005s against the configured PT20S/PT60S/PT120S; malformed key and malformed value each dead-lettered on first attempt with -Retry untouched, lag 0 and zero publication failures; no consumer assigned to any -Error topic. Tests: shared 12, measureeval 125, validation 153, 0 failures. Claude-Session: https://claude.ai/code/session_01AZubeEYCKfAmDCcgVTr18G
1 parent a2c38ce commit 62857e1

6 files changed

Lines changed: 306 additions & 17 deletions

File tree

‎Java/measureeval/src/main/java/com/lantanagroup/link/measureeval/configs/KafkaConfig.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -255,6 +255,14 @@ public RetryTopicConfiguration measureEvalRetryTopics(@Qualifier("compressedKafk
255255
.notRetryOn(DeserializationException.class)
256256
.useSingleTopicForSameIntervals()
257257
.doNotAutoCreateRetryTopics()
258+
// Keep the DLT in the destination chain — it is the routing target for container-thread
259+
// poison (see notRetryOn above), so removing it with doNotConfigureDlt() would leave a
260+
// malformed record with no destination and drop it instead of preserving it.
261+
// Only suppress its *listener*: nothing consumes -Error (there is no @DltHandler), so the
262+
// container it would otherwise start just re-reads each dead letter, fails again on the
263+
// same bytes, and commits the offset — which both doubles the error logging for poison
264+
// and advances the group past records that dead-letter replay will need to re-read.
265+
.autoStartDltHandler(false)
258266
.create(template);
259267
}
260268

‎Java/measureeval/src/test/java/com/lantanagroup/link/measureeval/configs/RetryTopicRecovererLoggingTest.java‎

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
package com.lantanagroup.link.measureeval.configs;
2+
3+
import ch.qos.logback.classic.Level;
4+
import ch.qos.logback.classic.Logger;
5+
import ch.qos.logback.classic.spi.ILoggingEvent;
6+
import ch.qos.logback.core.read.ListAppender;
7+
import com.lantanagroup.link.shared.config.KafkaRetryConfig;
8+
import com.lantanagroup.link.shared.exceptions.ValidationException;
9+
import com.lantanagroup.link.shared.kafka.RetryTopicRecoverer;
10+
import com.lantanagroup.link.shared.kafka.RetryTopicRecovererFactory;
11+
import org.apache.kafka.clients.consumer.ConsumerRecord;
12+
import org.apache.kafka.clients.producer.ProducerRecord;
13+
import org.apache.kafka.common.Node;
14+
import org.apache.kafka.common.PartitionInfo;
15+
import org.junit.jupiter.api.AfterEach;
16+
import org.junit.jupiter.api.BeforeEach;
17+
import org.junit.jupiter.api.Test;
18+
import org.mockito.ArgumentCaptor;
19+
import org.slf4j.LoggerFactory;
20+
import org.springframework.kafka.core.KafkaTemplate;
21+
import org.springframework.kafka.retrytopic.RetryTopicHeaders;
22+
23+
import java.nio.ByteBuffer;
24+
import java.time.Duration;
25+
import java.util.Collections;
26+
import java.util.List;
27+
import java.util.concurrent.CompletableFuture;
28+
29+
import static org.junit.jupiter.api.Assertions.*;
30+
import static org.mockito.ArgumentMatchers.any;
31+
import static org.mockito.ArgumentMatchers.anyString;
32+
import static org.mockito.Mockito.*;
33+
34+
/**
35+
* The recoverer's log line must describe what actually happened to the record.
36+
*
37+
* <p>{@link RetryTopicRecoverer} emits the log, but the destination is chosen by the resolver built in
38+
* {@link RetryTopicRecovererFactory}, which considers three reasons to dead-letter: retries disabled,
39+
* poison exception, attempts exhausted. A log derived from only one of those three reports a retry that
40+
* never happens for the other two — the record is already on {@code -Error} while the log says
41+
* "Retry attempt 1/N", and no dead-letter line is ever emitted.</p>
42+
*
43+
* <p>These tests drive the real production wiring and assert the log against the topic the record was
44+
* actually published to, so the two can never drift apart again.</p>
45+
*/
46+
class RetryTopicRecovererLoggingTest {
47+
48+
private static final String RETRY_TOPIC = "ResourcesNormalized-Retry";
49+
private static final String ERROR_TOPIC = "ResourcesNormalized-Error";
50+
51+
private Logger recovererLogger;
52+
private ListAppender<ILoggingEvent> appender;
53+
private Level originalLevel;
54+
55+
@BeforeEach
56+
void attachAppender() {
57+
recovererLogger = (Logger) LoggerFactory.getLogger(RetryTopicRecoverer.class);
58+
originalLevel = recovererLogger.getLevel();
59+
recovererLogger.setLevel(Level.TRACE);
60+
appender = new ListAppender<>();
61+
appender.start();
62+
recovererLogger.addAppender(appender);
63+
}
64+
65+
@AfterEach
66+
void detachAppender() {
67+
recovererLogger.detachAppender(appender);
68+
appender.stop();
69+
recovererLogger.setLevel(originalLevel);
70+
}
71+
72+
/** What the recoverer did with a record: where it went, and what it said about it. */
73+
private record Outcome(String topic, ILoggingEvent log) {
74+
String message() {
75+
return log.getFormattedMessage();
76+
}
77+
}
78+
79+
private static ConsumerRecord<String, String> record() {
80+
return new ConsumerRecord<>("ResourcesNormalized", 0, 0L, "key", "value");
81+
}
82+
83+
/** Pre-stamp the attempts header as if {@code attempts} retries already occurred. */
84+
private static void stampAttempts(ConsumerRecord<?, ?> record, int attempts) {
85+
record.headers().add(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS,
86+
ByteBuffer.allocate(Integer.BYTES).putInt(attempts).array());
87+
}
88+
89+
private static KafkaRetryConfig retryConfig(int maxAttempts, boolean disableRetryConsumer) {
90+
KafkaRetryConfig config = new KafkaRetryConfig();
91+
// effectiveMaxAttempts() == size + 1, so (maxAttempts - 1) delays yields exactly maxAttempts attempts.
92+
config.setConsumerRetryDuration(Collections.nCopies(Math.max(maxAttempts - 1, 0), Duration.ofSeconds(1)));
93+
config.setDisableRetryConsumer(disableRetryConsumer);
94+
return config;
95+
}
96+
97+
/**
98+
* Drives a failed record through the real {@link RetryTopicRecovererFactory#create} wiring with the
99+
* production poison set, and captures both the topic the resolver published to and the single log
100+
* line the recoverer emitted.
101+
*/
102+
@SuppressWarnings("unchecked")
103+
private Outcome run(KafkaRetryConfig config, ConsumerRecord<String, String> record, Exception exception) {
104+
KafkaTemplate<Object, Object> template = mock(KafkaTemplate.class);
105+
when(template.partitionsFor(anyString())).thenAnswer(invocation ->
106+
List.of(new PartitionInfo(invocation.getArgument(0), 0, null, new Node[0], new Node[0])));
107+
when(template.send(any(ProducerRecord.class)))
108+
.thenReturn(CompletableFuture.completedFuture(null));
109+
110+
RetryTopicRecoverer recoverer = RetryTopicRecovererFactory.create(
111+
template, RETRY_TOPIC, ERROR_TOPIC, config, KafkaConfig.NON_RETRYABLE);
112+
recoverer.accept(record, exception);
113+
114+
ArgumentCaptor<ProducerRecord<Object, Object>> captor = ArgumentCaptor.forClass(ProducerRecord.class);
115+
verify(template).send(captor.capture());
116+
117+
assertEquals(1, appender.list.size(),
118+
"recoverer must emit exactly one log line per failure, got: " + appender.list);
119+
return new Outcome(captor.getValue().topic(), appender.list.get(0));
120+
}
121+
122+
@Test
123+
void poisonException_isNotLoggedAsARetry() {
124+
Outcome outcome = run(retryConfig(3, false), record(), new ValidationException("Query Type is null."));
125+
126+
assertEquals(ERROR_TOPIC, outcome.topic(), "precondition: poison must route to -Error");
127+
assertFalse(outcome.message().contains("Retry attempt"),
128+
"record was dead-lettered but logged as a retry: " + outcome.message());
129+
assertEquals(Level.WARN, outcome.log().getLevel(),
130+
"dead-lettering must be logged at WARN, not buried at INFO: " + outcome.message());
131+
}
132+
133+
@Test
134+
void disabledRetryConsumer_isNotLoggedAsARetry() {
135+
Outcome outcome = run(retryConfig(3, true), record(), new RuntimeException("transient boom"));
136+
137+
assertEquals(ERROR_TOPIC, outcome.topic(), "precondition: disabled retries must route to -Error");
138+
assertFalse(outcome.message().contains("Retry attempt"),
139+
"record was dead-lettered but logged as a retry: " + outcome.message());
140+
assertEquals(Level.WARN, outcome.log().getLevel(),
141+
"dead-lettering must be logged at WARN, not buried at INFO: " + outcome.message());
142+
}
143+
144+
@Test
145+
void exhaustedAttempts_isLoggedAsADeadLetter() {
146+
ConsumerRecord<String, String> record = record();
147+
stampAttempts(record, 2);
148+
149+
Outcome outcome = run(retryConfig(3, false), record, new RuntimeException("still failing"));
150+
151+
assertEquals(ERROR_TOPIC, outcome.topic(), "precondition: exhausted attempts must route to -Error");
152+
assertFalse(outcome.message().contains("Retry attempt"),
153+
"record was dead-lettered but logged as a retry: " + outcome.message());
154+
assertEquals(Level.WARN, outcome.log().getLevel(),
155+
"dead-lettering must be logged at WARN: " + outcome.message());
156+
}
157+
158+
@Test
159+
void transientFailure_withAttemptsRemaining_isLoggedAsARetry() {
160+
Outcome outcome = run(retryConfig(3, false), record(), new RuntimeException("transient boom"));
161+
162+
assertEquals(RETRY_TOPIC, outcome.topic(), "precondition: transient failure must route to -Retry");
163+
assertTrue(outcome.message().contains("Retry attempt"),
164+
"record was retried but not logged as a retry: " + outcome.message());
165+
}
166+
}

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

Lines changed: 49 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -10,25 +10,61 @@
1010

1111
import java.math.BigInteger;
1212
import java.nio.ByteBuffer;
13+
import java.util.function.BiFunction;
1314
import java.util.function.IntToLongFunction;
1415

1516
public class RetryTopicRecoverer implements ConsumerRecordRecoverer {
1617

1718
private static final Logger logger = LoggerFactory.getLogger(RetryTopicRecoverer.class);
1819

20+
/**
21+
* What is about to happen to a failed record. The destination resolver and this recoverer's log line
22+
* are both derived from this single value, so the log can never claim a retry that the resolver has
23+
* already decided against.
24+
*/
25+
public enum Decision {
26+
/** Republished to the {@code -Retry} topic for another attempt. */
27+
RETRY,
28+
/** Dead-lettered: retries are switched off by configuration. */
29+
DLT_DISABLED,
30+
/** Dead-lettered: the exception can never succeed on retry. */
31+
DLT_POISON,
32+
/** Dead-lettered: no attempts remain. */
33+
DLT_EXHAUSTED
34+
}
35+
1936
private final int maxAttempts;
2037
private final IntToLongFunction backoffMsForAttempt;
2138
private final DeadLetterPublishingRecoverer delegate;
39+
private final BiFunction<ConsumerRecord<?, ?>, Exception, Decision> decide;
2240

2341
/**
2442
* @param maxAttempts attempt at which a record is exhausted (used for logging)
2543
* @param backoffMsForAttempt maps a 1-based attempt number to the backoff delay in millis
2644
* @param delegate publishes the record to the resolved {@code -Retry}/{@code -Error} topic
45+
* @param decide the same decision the delegate's destination resolver applies; evaluated
46+
* after the attempts header is stamped so both see the identical count
2747
*/
28-
public RetryTopicRecoverer(int maxAttempts, IntToLongFunction backoffMsForAttempt, DeadLetterPublishingRecoverer delegate) {
48+
public RetryTopicRecoverer(int maxAttempts,
49+
IntToLongFunction backoffMsForAttempt,
50+
DeadLetterPublishingRecoverer delegate,
51+
BiFunction<ConsumerRecord<?, ?>, Exception, Decision> decide) {
2952
this.maxAttempts = maxAttempts;
3053
this.backoffMsForAttempt = backoffMsForAttempt;
3154
this.delegate = delegate;
55+
this.decide = decide;
56+
}
57+
58+
/**
59+
* Attempts-only variant for callers with no poison classification or disable flag — where exhausting
60+
* the attempt budget genuinely is the only reason a record stops retrying. Prefer
61+
* {@link RetryTopicRecovererFactory#create} in production so the log reflects every routing reason.
62+
*/
63+
public RetryTopicRecoverer(int maxAttempts, IntToLongFunction backoffMsForAttempt, DeadLetterPublishingRecoverer delegate) {
64+
this(maxAttempts, backoffMsForAttempt, delegate,
65+
(record, exception) -> currentAttempts(record.headers()) >= maxAttempts
66+
? Decision.DLT_EXHAUSTED
67+
: Decision.RETRY);
3268
}
3369

3470
@Override
@@ -47,12 +83,19 @@ public void accept(ConsumerRecord<?, ?> record, Exception exception) {
4783
}
4884

4985
// Exception headers are stamped by the DeadLetterPublishingRecoverer delegate on publish.
50-
if (attempt >= maxAttempts) {
51-
logger.warn("Max retry attempts ({}) reached for topic [{}]. Routing to DLT.",
52-
maxAttempts, record.topic());
53-
} else {
54-
logger.info("Retry attempt {}/{} for topic [{}].",
86+
// Evaluated here, after the attempts header is stamped, so this sees the same count the
87+
// delegate's destination resolver will read a moment later.
88+
switch (decide.apply(record, exception)) {
89+
case RETRY -> logger.info("Retry attempt {}/{} for topic [{}].",
5590
attempt, maxAttempts, record.topic());
91+
case DLT_EXHAUSTED -> logger.warn("Max retry attempts ({}) reached for topic [{}]. Routing to DLT.",
92+
maxAttempts, record.topic());
93+
case DLT_POISON -> logger.warn(
94+
"Non-retryable exception [{}] for topic [{}]. Routing to DLT without retrying.",
95+
exception == null ? "unknown" : exception.getClass().getName(), record.topic());
96+
case DLT_DISABLED -> logger.warn(
97+
"Retry consumer is disabled. Routing record from topic [{}] to DLT without retrying.",
98+
record.topic());
5699
}
57100

58101
delegate.accept(record, exception);

‎Java/shared/src/main/java/com/lantanagroup/link/shared/kafka/RetryTopicRecovererFactory.java‎

Lines changed: 24 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -37,16 +37,30 @@ public static RetryTopicRecoverer create(
3737
KafkaRetryConfig config,
3838
Set<Class<? extends Throwable>> nonRetryable) {
3939

40-
BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> resolver =
40+
// The single routing decision. Both the destination resolver below and the recoverer's log line
41+
// are derived from this, so a record can never be dead-lettered while the log reports a retry.
42+
// Route straight to the error topic when retries are disabled, the message is poison (malformed
43+
// content / deserialization can never succeed), or attempts are exhausted.
44+
BiFunction<ConsumerRecord<?, ?>, Exception, RetryTopicRecoverer.Decision> decide =
4145
(record, exception) -> {
46+
if (config.isDisableRetryConsumer()) {
47+
return RetryTopicRecoverer.Decision.DLT_DISABLED;
48+
}
49+
if (isNonRetryable(exception, nonRetryable)) {
50+
return RetryTopicRecoverer.Decision.DLT_POISON;
51+
}
4252
int attempt = RetryTopicRecoverer.currentAttempts(record.headers());
43-
// Route straight to the error topic when retries are disabled, the message is poison
44-
// (malformed content / deserialization can never succeed), or attempts are exhausted.
45-
String target = (config.isDisableRetryConsumer()
46-
|| isNonRetryable(exception, nonRetryable)
47-
|| attempt >= config.effectiveMaxAttempts())
48-
? errorTopic
49-
: retryTopic;
53+
if (attempt >= config.effectiveMaxAttempts()) {
54+
return RetryTopicRecoverer.Decision.DLT_EXHAUSTED;
55+
}
56+
return RetryTopicRecoverer.Decision.RETRY;
57+
};
58+
59+
BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> resolver =
60+
(record, exception) -> {
61+
String target = decide.apply(record, exception) == RetryTopicRecoverer.Decision.RETRY
62+
? retryTopic
63+
: errorTopic;
5064
return new TopicPartition(target, record.partition());
5165
};
5266

@@ -55,7 +69,8 @@ public static RetryTopicRecoverer create(
5569
return new RetryTopicRecoverer(
5670
config.effectiveMaxAttempts(),
5771
config::backoffMsForAttempt,
58-
delegate);
72+
delegate,
73+
decide);
5974
}
6075

6176
/**

‎Java/validation/src/main/java/com/lantanagroup/link/validation/configs/KafkaConfig.java‎

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,11 @@ public Deserializer<?> valueDeserializer(ObjectMapper objectMapper) {
9898
public Serializer<?> keySerializer(ObjectMapper objectMapper) {
9999
Map<Class<?>, Serializer<?>> serializers = Map.of(
100100
String.class, new StringSerializer(),
101-
ReadyForValidation.Key.class, getJsonSerializer(objectMapper, ReadyForValidation.Key.class)
101+
ReadyForValidation.Key.class, getJsonSerializer(objectMapper, ReadyForValidation.Key.class),
102+
// A record whose key fails to deserialize carries raw bytes, and those bytes are what the
103+
// dead-letter publisher must write. Without this delegate the publication throws and the
104+
// container redelivers the record forever, blocking the partition.
105+
byte[].class, new ByteArraySerializer()
102106
);
103107
return new DelegatingByTypeSerializer(serializers);
104108
}
@@ -108,7 +112,12 @@ public Serializer<?> valueSerializer(ObjectMapper objectMapper) {
108112
Map<Class<?>, Serializer<?>> serializers = Map.of(
109113
String.class, new StringSerializer(),
110114
ValidationComplete.class, getJsonSerializer(objectMapper, ValidationComplete.class),
111-
ReadyForValidation.class, getJsonSerializer(objectMapper, ReadyForValidation.class)
115+
ReadyForValidation.class, getJsonSerializer(objectMapper, ReadyForValidation.class),
116+
// A record that fails deserialization reaches the dead-letter publisher as the raw byte[]
117+
// that could not be parsed — there is no bound type to serialize. Without this delegate the
118+
// publication throws, DefaultErrorHandler cannot recover, and the container redelivers the
119+
// record forever, blocking the partition and stranding every message behind it.
120+
byte[].class, new ByteArraySerializer()
112121
);
113122
return new DelegatingByTypeSerializer(serializers);
114123
}
@@ -216,6 +225,14 @@ public RetryTopicConfiguration validationRetryTopics(@Qualifier("compressedKafka
216225
.notRetryOn(DeserializationException.class)
217226
.useSingleTopicForSameIntervals()
218227
.doNotAutoCreateRetryTopics()
228+
// Keep the DLT in the destination chain — it is the routing target for container-thread
229+
// poison (see notRetryOn above), so removing it with doNotConfigureDlt() would leave a
230+
// malformed record with no destination and drop it instead of preserving it.
231+
// Only suppress its *listener*: nothing consumes -Error (there is no @DltHandler), so the
232+
// container it would otherwise start just re-reads each dead letter, fails again on the
233+
// same bytes, and commits the offset — which both doubles the error logging for poison
234+
// and advances the group past records that dead-letter replay will need to re-read.
235+
.autoStartDltHandler(false)
219236
.create(template);
220237
}
221238

0 commit comments

Comments
 (0)