Skip to content

Commit 1f5dd35

Browse files
committed
review fix
* @author classes. * fix adoc * poblish `AfterRollbackProcessor` * javadoc at `ContainerProperties`
1 parent d2484f4 commit 1f5dd35

File tree

10 files changed

+31
-6
lines changed

10 files changed

+31
-6
lines changed

Diff for: spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/annotation-error-handling.adoc

+1-2
Original file line numberDiff line numberDiff line change
@@ -454,8 +454,7 @@ See xref:kafka/annotation-error-handling.adoc#error-handlers[Container Error Han
454454
Starting with version 3.2, Recovery can now recover (skip) entire batch of records that keeps failing.
455455
Set `ContainerProperties.setBatchRecoverAfterRollback(true)` to enable this feature.
456456

457-
IMPORTANT: Default behavior, recovery is not possible with a batch listener,
458-
since the framework has no knowledge about which record in the batch keeps failing.
457+
IMPORTANT: Default behavior, recovery is not possible with a batch listener, since the framework has no knowledge about which record in the batch keeps failing.
459458
In such cases, the application listener must handle a record that keeps failing.
460459

461460
See also xref:kafka/annotation-error-handling.adoc#dead-letters[Publishing Dead-letter Records].

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/AfterRollbackProcessor.java

+2
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
* @param <V> the value type.
3636
*
3737
* @author Gary Russell
38+
* @author Wang Zhiyang
3839
*
3940
* @since 1.3.5
4041
*
@@ -80,6 +81,7 @@ void process(List<ConsumerRecord<K, V>> records, Consumer<K, V> consumer,
8081
default void processBatch(ConsumerRecords<K, V> records, List<ConsumerRecord<K, V>> recordList,
8182
Consumer<K, V> consumer, MessageListenerContainer container, Exception exception,
8283
boolean recoverable, ContainerProperties.EOSMode eosMode) {
84+
8385
process(recordList, consumer, container, exception, recoverable, eosMode);
8486
}
8587

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/BatchListenerFailedException.java

+1
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
* failed.
2727
*
2828
* @author Gary Russell
29+
* @author Wang Zhiyang
2930
* @since 2.5
3031
*
3132
*/

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java

+11
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@
5151
* @author Johnny Lim
5252
* @author Lukasz Kaminski
5353
* @author Kyuhyeok Park
54+
* @author Wang Zhiyang
5455
*/
5556
public class ContainerProperties extends ConsumerProperties {
5657

@@ -545,10 +546,20 @@ public void setTransactionManager(@Nullable PlatformTransactionManager transacti
545546
this.transactionManager = transactionManager;
546547
}
547548

549+
/**
550+
* Recover batch records after rollback if true.
551+
* @return true to recover.
552+
* @since 3.2
553+
*/
548554
public boolean isBatchRecoverAfterRollback() {
549555
return this.batchRecoverAfterRollback;
550556
}
551557

558+
/**
559+
* enable the batch recover after rollback.
560+
* @param batchRecoverAfterRollback the batchRecoverAfterRollback to set.
561+
* @since 3.2
562+
*/
552563
public void setBatchRecoverAfterRollback(boolean batchRecoverAfterRollback) {
553564
this.batchRecoverAfterRollback = batchRecoverAfterRollback;
554565
}

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java

+1
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@
5050
*
5151
* @author Gary Russell
5252
* @author Francois Rosiere
53+
* @author Wang Zhiyang
5354
*
5455
* @since 1.3.5
5556
*

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

+10-4
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,7 @@
161161
* @author Francois Rosiere
162162
* @author Daniel Gentes
163163
* @author Soby Chacko
164+
* @author Wang Zhiyang
164165
*/
165166
public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
166167
extends AbstractMessageListenerContainer<K, V> implements ConsumerPauseResumeEventPublisher {
@@ -2204,10 +2205,15 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
22042205
});
22052206
}
22062207
else {
2207-
afterRollbackProcessorToUse.processBatch(records,
2208-
Objects.requireNonNullElseGet(recordList, () -> createRecordList(records)), this.consumer,
2209-
KafkaMessageListenerContainer.this.thisOrParentContainer, e,
2210-
this.wantsBatchRecoverAfterRollback, this.eosMode);
2208+
try {
2209+
afterRollbackProcessorToUse.processBatch(records,
2210+
Objects.requireNonNullElseGet(recordList, () -> createRecordList(records)), this.consumer,
2211+
KafkaMessageListenerContainer.this.thisOrParentContainer, e,
2212+
this.wantsBatchRecoverAfterRollback, this.eosMode);
2213+
}
2214+
catch (Exception ex) {
2215+
this.logger.error(ex, "AfterRollbackProcessor threw exception");
2216+
}
22112217
}
22122218
}
22132219

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java

+1
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
* @author Gary Russell
3232
* @author Francois Rosiere
3333
* @author Antonio Tomac
34+
* @author Wang Zhiyang
3435
* @since 2.0
3536
*
3637
*/

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/SeekUtils.java

+2
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
*
4646
* @author Gary Russell
4747
* @author Francois Rosiere
48+
* @author Wang Zhiyang
4849
* @since 2.2
4950
*
5051
*/
@@ -138,6 +139,7 @@ public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?>
138139
* @param records the records.
139140
* @param consumer the consumer.
140141
* @param logger a {@link LogAccessor} for seek errors.
142+
* @since 3.2
141143
*/
142144
public static void doSeeksToBegin(List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
143145
LogAccessor logger) {

Diff for: spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessorTests.java

+1
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@
5353
/**
5454
* @author Gary Russell
5555
* @author Francois Rosiere
56+
* @author Wang Zhiyang
5657
* @since 2.3.1
5758
*
5859
*/

Diff for: spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java

+1
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,7 @@
107107
/**
108108
* @author Gary Russell
109109
* @author Artem Bilan
110+
* @author Wang Zhiyang
110111
*
111112
* @since 1.3
112113
*

0 commit comments

Comments
 (0)