Skip to content

Commit 457e63a

Browse files
garyrussellartembilan
garyrussell
authored and
artembilan
committed
GH-1260: Add exception classification to DARP
Resolves #1260 Refactor common code with `SeekToCurrentErrorHandler` into `FailedRecordProcessor` super class. * Polishing to address PR comment
1 parent 37a1b5b commit 457e63a

File tree

5 files changed

+364
-171
lines changed

5 files changed

+364
-171
lines changed

‎spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java

Lines changed: 11 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -20,13 +20,11 @@
2020
import java.util.List;
2121
import java.util.function.BiConsumer;
2222

23-
import org.apache.commons.logging.LogFactory;
2423
import org.apache.kafka.clients.consumer.Consumer;
2524
import org.apache.kafka.clients.consumer.ConsumerRecord;
2625
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
2726
import org.apache.kafka.common.TopicPartition;
2827

29-
import org.springframework.core.log.LogAccessor;
3028
import org.springframework.kafka.core.KafkaTemplate;
3129
import org.springframework.kafka.support.SeekUtils;
3230
import org.springframework.lang.Nullable;
@@ -49,14 +47,7 @@
4947
* @since 1.3.5
5048
*
5149
*/
52-
public class DefaultAfterRollbackProcessor<K, V> implements AfterRollbackProcessor<K, V> {
53-
54-
private static final LogAccessor LOGGER =
55-
new LogAccessor(LogFactory.getLog(DefaultAfterRollbackProcessor.class));
56-
57-
private final FailedRecordTracker failureTracker;
58-
59-
private boolean commitRecovered;
50+
public class DefaultAfterRollbackProcessor<K, V> extends FailedRecordProcessor implements AfterRollbackProcessor<K, V> {
6051

6152
private KafkaTemplate<K, V> kafkaTemplate;
6253

@@ -126,7 +117,8 @@ public DefaultAfterRollbackProcessor(BiConsumer<ConsumerRecord<?, ?>, Exception>
126117
public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer,
127118
int maxFailures) {
128119

129-
this.failureTracker = new FailedRecordTracker(recoverer, new FixedBackOff(0L, maxFailures - 1), LOGGER);
120+
// Remove super CTOR when this is removed.
121+
super(recoverer, maxFailures);
130122
}
131123

132124
/**
@@ -139,16 +131,17 @@ public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>,
139131
public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer,
140132
BackOff backOff) {
141133

142-
this.failureTracker = new FailedRecordTracker(recoverer, backOff, LOGGER);
134+
super(recoverer, backOff);
143135
}
144136

145137
@SuppressWarnings({ "unchecked", "rawtypes" })
146138
@Override
147139
public void process(List<ConsumerRecord<K, V>> records, Consumer<K, V> consumer, Exception exception,
148140
boolean recoverable) {
149141

150-
if (SeekUtils.doSeeks(((List) records), consumer, exception, recoverable, this.failureTracker::skip, LOGGER)
151-
&& this.kafkaTemplate != null && this.kafkaTemplate.isTransactional()) {
142+
if (SeekUtils.doSeeks(((List) records), consumer, exception, recoverable,
143+
getSkipPredicate((List) records, exception), LOGGER)
144+
&& isCommitRecovered() && this.kafkaTemplate != null && this.kafkaTemplate.isTransactional()) {
152145
ConsumerRecord<K, V> skipped = records.get(0);
153146
this.kafkaTemplate.sendOffsetsToTransaction(
154147
Collections.singletonMap(new TopicPartition(skipped.topic(), skipped.partition()),
@@ -158,10 +151,11 @@ public void process(List<ConsumerRecord<K, V>> records, Consumer<K, V> consumer,
158151

159152
@Override
160153
public boolean isProcessInTransaction() {
161-
return this.commitRecovered;
154+
return isCommitRecovered();
162155
}
163156

164157
/**
158+
* {@inheritDoc}
165159
* Set to true to and the container will run the
166160
* {@link #process(List, Consumer, Exception, boolean)} method in a transaction and,
167161
* if a record is skipped and recovered, we will send its offset to the transaction.
@@ -172,8 +166,9 @@ public boolean isProcessInTransaction() {
172166
* @see #process(List, Consumer, Exception, boolean)
173167
* @see #setKafkaTemplate(KafkaTemplate)
174168
*/
169+
@Override
175170
public void setCommitRecovered(boolean commitRecovered) {
176-
this.commitRecovered = commitRecovered;
171+
super.setCommitRecovered(commitRecovered); // NOSONAR enhanced javadoc
177172
}
178173

179174
/**
@@ -187,9 +182,4 @@ public void setKafkaTemplate(KafkaTemplate<K, V> kafkaTemplate) {
187182
this.kafkaTemplate = kafkaTemplate;
188183
}
189184

190-
@Override
191-
public void clearThreadState() {
192-
this.failureTracker.clearThreadState();
193-
}
194-
195185
}

‎spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java

Lines changed: 223 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,223 @@
1+
/*
2+
* Copyright 2019 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.kafka.listener;
18+
19+
import java.util.HashMap;
20+
import java.util.List;
21+
import java.util.Map;
22+
import java.util.function.BiConsumer;
23+
import java.util.function.BiPredicate;
24+
25+
import org.apache.commons.logging.LogFactory;
26+
import org.apache.kafka.clients.consumer.ConsumerRecord;
27+
28+
import org.springframework.classify.BinaryExceptionClassifier;
29+
import org.springframework.core.log.LogAccessor;
30+
import org.springframework.kafka.support.serializer.DeserializationException;
31+
import org.springframework.lang.Nullable;
32+
import org.springframework.messaging.converter.MessageConversionException;
33+
import org.springframework.messaging.handler.invocation.MethodArgumentResolutionException;
34+
import org.springframework.util.Assert;
35+
import org.springframework.util.backoff.BackOff;
36+
import org.springframework.util.backoff.FixedBackOff;
37+
38+
/**
39+
* Common super class for classes that deal with failing to consume a consumer record.
40+
*
41+
* @author Gary Russell
42+
* @since 2.3.1
43+
*
44+
*/
45+
public abstract class FailedRecordProcessor {
46+
47+
private static final BiPredicate<ConsumerRecord<?, ?>, Exception> ALWAYS_SKIP_PREDICATE = (r, e) -> true;
48+
49+
private static final BiPredicate<ConsumerRecord<?, ?>, Exception> NEVER_SKIP_PREDICATE = (r, e) -> false;
50+
51+
protected static final LogAccessor LOGGER =
52+
new LogAccessor(LogFactory.getLog(SeekToCurrentErrorHandler.class)); // NOSONAR
53+
54+
private final FailedRecordTracker failureTracker;
55+
56+
private BinaryExceptionClassifier classifier;
57+
58+
private boolean commitRecovered;
59+
60+
protected FailedRecordProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff) {
61+
this.failureTracker = new FailedRecordTracker(recoverer, backOff, LOGGER);
62+
this.classifier = configureDefaultClassifier();
63+
}
64+
65+
/**
66+
* TODO: remove when the deprecated dependent CTORs are removed.
67+
* @param recoverer the recoverer.
68+
* @param maxFailures the max failures.
69+
*/
70+
FailedRecordProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, int maxFailures) {
71+
this.failureTracker = new FailedRecordTracker(recoverer, new FixedBackOff(0L, maxFailures - 1), LOGGER);
72+
this.classifier = configureDefaultClassifier();
73+
}
74+
75+
/**
76+
* Return the exception classifier.
77+
* @return the classifier.
78+
*/
79+
protected BinaryExceptionClassifier getClassifier() {
80+
return this.classifier;
81+
}
82+
83+
/**
84+
* Set an exception classifications to determine whether the exception should cause a retry
85+
* (until exhaustion) or not. If not, we go straight to the recoverer. By default,
86+
* the following exceptions will not be retried:
87+
* <ul>
88+
* <li>{@link DeserializationException}</li>
89+
* <li>{@link MessageConversionException}</li>
90+
* <li>{@link MethodArgumentResolutionException}</li>
91+
* <li>{@link NoSuchMethodException}</li>
92+
* <li>{@link ClassCastException}</li>
93+
* </ul>
94+
* All others will be retried.
95+
* When calling this method, the defaults will not be applied.
96+
* @param classifications the classifications.
97+
* @param defaultValue whether or not to retry non-matching exceptions.
98+
* @see BinaryExceptionClassifier#BinaryExceptionClassifier(Map, boolean)
99+
*/
100+
public void setClassifications(Map<Class<? extends Throwable>, Boolean> classifications, boolean defaultValue) {
101+
Assert.notNull(classifications, "'classifications' + cannot be null");
102+
this.classifier = new ExtendedBinaryExceptionClassifier(classifications, defaultValue);
103+
}
104+
105+
protected void setClassifier(BinaryExceptionClassifier classifier) {
106+
this.classifier = classifier;
107+
}
108+
109+
/**
110+
* Whether the offset for a recovered record should be committed.
111+
* @return true to commit recovered record offsets.
112+
*/
113+
protected boolean isCommitRecovered() {
114+
return this.commitRecovered;
115+
}
116+
117+
/**
118+
* Set to true to commit the offset for a recovered record.
119+
* @param commitRecovered true to commit.
120+
*/
121+
public void setCommitRecovered(boolean commitRecovered) {
122+
this.commitRecovered = commitRecovered;
123+
}
124+
125+
/**
126+
* Add an exception type to the default list; if and only if an external classifier
127+
* has not been provided. By default, the following exceptions will not be retried:
128+
* <ul>
129+
* <li>{@link DeserializationException}</li>
130+
* <li>{@link MessageConversionException}</li>
131+
* <li>{@link MethodArgumentResolutionException}</li>
132+
* <li>{@link NoSuchMethodException}</li>
133+
* <li>{@link ClassCastException}</li>
134+
* </ul>
135+
* All others will be retried.
136+
* @param exceptionType the exception type.
137+
* @see #removeNotRetryableException(Class)
138+
* @see #setClassifications(Map, boolean)
139+
*/
140+
public void addNotRetryableException(Class<? extends Exception> exceptionType) {
141+
Assert.isTrue(this.classifier instanceof ExtendedBinaryExceptionClassifier,
142+
"Cannot add exception types to a supplied classifier");
143+
((ExtendedBinaryExceptionClassifier) this.classifier).getClassified().put(exceptionType, false);
144+
}
145+
146+
/**
147+
* Remove an exception type from the configured list; if and only if an external
148+
* classifier has not been provided. By default, the following exceptions will not be
149+
* retried:
150+
* <ul>
151+
* <li>{@link DeserializationException}</li>
152+
* <li>{@link MessageConversionException}</li>
153+
* <li>{@link MethodArgumentResolutionException}</li>
154+
* <li>{@link NoSuchMethodException}</li>
155+
* <li>{@link ClassCastException}</li>
156+
* </ul>
157+
* All others will be retried.
158+
* @param exceptionType the exception type.
159+
* @return true if the removal was successful.
160+
* @see #addNotRetryableException(Class)
161+
* @see #setClassifications(Map, boolean)
162+
*/
163+
public boolean removeNotRetryableException(Class<? extends Exception> exceptionType) {
164+
Assert.isTrue(this.classifier instanceof ExtendedBinaryExceptionClassifier,
165+
"Cannot remove exception types from a supplied classifier");
166+
return ((ExtendedBinaryExceptionClassifier) this.classifier).getClassified().remove(exceptionType);
167+
}
168+
169+
protected BiPredicate<ConsumerRecord<?, ?>, Exception> getSkipPredicate(List<ConsumerRecord<?, ?>> records,
170+
Exception thrownException) {
171+
172+
if (getClassifier().classify(thrownException)) {
173+
return this.failureTracker::skip;
174+
}
175+
else {
176+
try {
177+
this.failureTracker.getRecoverer().accept(records.get(0), thrownException);
178+
}
179+
catch (Exception ex) {
180+
LOGGER.error(ex, () -> "Recovery of record (" + records.get(0) + ") failed");
181+
return NEVER_SKIP_PREDICATE;
182+
}
183+
return ALWAYS_SKIP_PREDICATE;
184+
}
185+
}
186+
187+
public void clearThreadState() {
188+
this.failureTracker.clearThreadState();
189+
}
190+
191+
private static BinaryExceptionClassifier configureDefaultClassifier() {
192+
Map<Class<? extends Throwable>, Boolean> classified = new HashMap<>();
193+
classified.put(DeserializationException.class, false);
194+
classified.put(MessageConversionException.class, false);
195+
classified.put(MethodArgumentResolutionException.class, false);
196+
classified.put(NoSuchMethodException.class, false);
197+
classified.put(ClassCastException.class, false);
198+
ExtendedBinaryExceptionClassifier defaultClassifier = new ExtendedBinaryExceptionClassifier(classified, true);
199+
return defaultClassifier;
200+
}
201+
202+
/**
203+
* Extended to provide visibility to the current classified exceptions.
204+
*
205+
* @author Gary Russell
206+
*
207+
*/
208+
@SuppressWarnings("serial")
209+
private static final class ExtendedBinaryExceptionClassifier extends BinaryExceptionClassifier {
210+
211+
ExtendedBinaryExceptionClassifier(Map<Class<? extends Throwable>, Boolean> typeMap, boolean defaultValue) {
212+
super(typeMap, defaultValue);
213+
setTraverseCauses(true);
214+
}
215+
216+
@Override
217+
protected Map<Class<? extends Throwable>, Boolean> getClassified() { // NOSONAR worthless override
218+
return super.getClassified();
219+
}
220+
221+
}
222+
223+
}

0 commit comments

Comments
 (0)