Fix bug at TransactionalContainerTests.testBatchListenerRecoverAfterRollbackProcessorCrash

* change CountDownLatch to 2, record foo and bar in the same throw exception, baz and qux count down latch.

Author: Wzy19930507

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

@@ -112,7 +112,7 @@
 @EmbeddedKafka(topics = { TransactionalContainerTests.topic1, TransactionalContainerTests.topic2,
 		TransactionalContainerTests.topic3, TransactionalContainerTests.topic3DLT, TransactionalContainerTests.topic4,
 		TransactionalContainerTests.topic5, TransactionalContainerTests.topic6, TransactionalContainerTests.topic7,
-		TransactionalContainerTests.topic8, TransactionalContainerTests.topic8DLT },
+		TransactionalContainerTests.topic8, TransactionalContainerTests.topic8DLT, TransactionalContainerTests.topic9},
 		brokerProperties = { "transaction.state.log.replication.factor=1", "transaction.state.log.min.isr=1" })
 public class TransactionalContainerTests {
 
@@ -957,7 +957,7 @@ public void testBatchListenerRecoverAfterRollbackProcessorCrash() throws Excepti
 		DefaultKafkaProducerFactory<Object, Object> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		pf.setTransactionIdPrefix("batchListener.noRetries.");
 		final KafkaTemplate<Object, Object> template = new KafkaTemplate<>(pf);
-		final CountDownLatch latch = new CountDownLatch(1);
+		final CountDownLatch latch = new CountDownLatch(2);
 		AtomicReference<String> data = new AtomicReference<>();
 		containerProps.setMessageListener((BatchMessageListener<Integer, String>) recordList -> {
 			for (ConsumerRecord<Integer, String> record : recordList) {