fix review and fix test bug at DefaultAfterRollbackProcessorTests

* add @test to DefaultAfterRollbackProcessorTests.testNoEarlyExitBackOff
* polish TransactionalContainerTests
* fix bug Tests at DefaultAfterRollbackProcessorTests method `testNoEarlyExitBackOff` and testEarlyExitBackOff

Author: Wzy19930507

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2023 the original author or authors.
+ * Copyright 2019-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -54,12 +54,12 @@
  * @author Gary Russell
  * @author Francois Rosiere
  * @author Wang Zhiyang
+ *
  * @since 2.3.1
  *
  */
 public class DefaultAfterRollbackProcessorTests {
 
-	@SuppressWarnings("deprecation")
 	@Test
 	void testClassifier() {
 		AtomicReference<ConsumerRecord<?, ?>> recovered = new AtomicReference<>();
@@ -142,10 +142,27 @@ void testBackOffNoBatchRecover() {
 		verify(backOff, times(3)).start();
 	}
 
+	@Test
+	void testEarlyExitBackOff() {
+		DefaultAfterRollbackProcessor<String, String> processor = new DefaultAfterRollbackProcessor<>(
+				new FixedBackOff(10_000, 1));
+		@SuppressWarnings("unchecked")
+		Consumer<String, String> consumer = mock(Consumer.class);
+		ConsumerRecord<String, String> record1 = new ConsumerRecord<>("foo", 0, 0L, "foo", "bar");
+		ConsumerRecord<String, String> record2 = new ConsumerRecord<>("foo", 1, 1L, "foo", "bar");
+		List<ConsumerRecord<String, String>> records = Arrays.asList(record1, record2);
+		IllegalStateException illegalState = new IllegalStateException();
+		MessageListenerContainer container = mock(MessageListenerContainer.class);
+		given(container.isRunning()).willReturn(false);
+		long t1 = System.currentTimeMillis();
+		processor.process(records, consumer, container, illegalState, true, EOSMode.V2);
+		assertThat(System.currentTimeMillis() < t1 + 5_000).isTrue();
+	}
+
 	@Test
 	void testNoEarlyExitBackOff() {
 		DefaultAfterRollbackProcessor<String, String> processor = new DefaultAfterRollbackProcessor<>(
-				new FixedBackOff(1, 200));
+				new FixedBackOff(200, 1));
 		@SuppressWarnings("unchecked")
 		Consumer<String, String> consumer = mock(Consumer.class);
 		ConsumerRecord<String, String> record1 = new ConsumerRecord<>("foo", 0, 0L, "foo", "bar");
@@ -156,7 +173,7 @@ void testNoEarlyExitBackOff() {
 		given(container.isRunning()).willReturn(true);
 		long t1 = System.currentTimeMillis();
 		processor.process(records, consumer, container, illegalState, true, EOSMode.V2);
-		assertThat(System.currentTimeMillis() >= t1 + 200);
+		assertThat(System.currentTimeMillis() >= t1 + 200).isTrue();
 	}
 
 }

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

@@ -788,8 +788,6 @@ public void accept(ConsumerRecord<?, ?> record, Consumer<?, ?> consumer, Excepti
 	@SuppressWarnings({ "unchecked"})
 	@Test
 	public void testBatchListenerMaxFailuresOnRecover() throws Exception {
-
-		logger.info("Start testBatchListenerMaxFailures");
 		String group = "groupInARBP2";
 		Map<String, Object> props = KafkaTestUtils.consumerProps(group, "false", embeddedKafka);
 		props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
@@ -904,7 +902,6 @@ public void accept(ConsumerRecord<?, ?> record, Consumer<?, ?> consumer, Excepti
 		verify(dlTemplate).sendOffsetsToTransaction(
 				eq(Collections.singletonMap(new TopicPartition(topic8, 0), new OffsetAndMetadata(4L))),
 				any(ConsumerGroupMetadata.class));
-		logger.info("Stop testBatchListenerMaxFailures");
 	}
 
 	@SuppressWarnings("unchecked")
@@ -973,7 +970,6 @@ public void testRollbackProcessorCrash() throws Exception {
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testBatchListenerRecoverAfterRollbackProcessorCrash() throws Exception {
-		logger.info("Start testBatchListenerRollbackNoRetries");
 		Map<String, Object> props = KafkaTestUtils.consumerProps("testBatchListenerRollbackNoRetries", "false", embeddedKafka);
 		props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
 		props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2);
@@ -1041,10 +1037,9 @@ public void accept(ConsumerRecord<?, ?> record, Consumer<?, ?> consumer, Excepti
 		container.stop();
 		pf.destroy();
 		assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue();
-		logger.info("Stop testRollbackNoRetries");
 	}
 
-	@SuppressWarnings({ "rawtypes", "unchecked", "deprecation" })
+	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
 	void testNoAfterRollbackWhenFenced() throws Exception {
 		Consumer consumer = mock(Consumer.class);