Restore STCEH Behavior After Previous Commit

spring-projects · Jun 23, 2021 · 924b8a8 · 924b8a8
1 parent bb3d3df
commit 924b8a8
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 6 deletions.
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java
@@ -109,10 +109,8 @@ public void setAckAfterHandle(boolean ackAfterHandle) {
 	public void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container) {
 
-		if (records != null) {
-			SeekUtils.seekOrRecover(thrownException, records, consumer, container, isCommitRecovered(),
-					getRecoveryStrategy(records, thrownException), this.logger, getLogLevel());
-		}
+		SeekUtils.seekOrRecover(thrownException, records, consumer, container, isCommitRecovered(),
+				getRecoveryStrategy(records, thrownException), this.logger, getLogLevel());
 	}
 
 }
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekUtils.java
@@ -185,7 +185,7 @@ public static void seekOrRecover(Exception thrownException, List<ConsumerRecord<
 	 * @param level the log level for the thrown exception after handling.
 	 * @since 2.7
 	 */
-	public static void seekOrRecover(Exception thrownException, List<ConsumerRecord<?, ?>> records,
+	public static void seekOrRecover(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container, boolean commitRecovered,
 			RecoveryStrategy recovery, LogAccessor logger, Level level) {
 
@@ -202,7 +202,7 @@ public static void seekOrRecover(Exception thrownException, List<ConsumerRecord<
 			}
 		}
 
-		if (!doSeeks(records, consumer, thrownException, true, recovery, container, logger)) {
+		if (records == null || !doSeeks(records, consumer, thrownException, true, recovery, container, logger)) {
 			throw new KafkaException("Seek to current after exception", level, thrownException);
 		}
 		if (commitRecovered) {