Fix KafkaMLCTests for proper CountDownLatch place

In the `KafkaMessageListenerContainerTests.invokeBatchInterceptorSuccessFailureOnRetry()`
we are expecting interaction with a `BatchInterceptor`, when its `success` or `failure` happens
already after the `MessageListener` call, but a `CountDownLatch` thread barrier in the test setup
is counted in the `MessageListener` leading to race condition when we exit from the barrier to
verification phase, but `BatchInterceptor.success()` has not been called yet
in the other thread.

* Fix `KafkaMessageListenerContainerTests.invokeBatchInterceptorSuccessFailureOnRetry()`
moving `CountDownLatch` to the `BatchInterceptor.success()` implementation.
Leave `MessageListener` just only with an `AtomicInteger` for retry attempts

Author: artembilan

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

@@ -4313,14 +4313,14 @@ public void invokeBatchInterceptorSuccessFailureOnRetry() throws Exception {
 		TopicPartitionOffset[] topicPartition = new TopicPartitionOffset[] {
 				new TopicPartitionOffset("test-topic", 0) };
 
-		CountDownLatch latch = new CountDownLatch(4); // 3 failures, 1 success
+		AtomicInteger attempts = new AtomicInteger(3); // 3 failures, 1 success
+		// Cannot be lambda: Mockito doesn't mock final classes
 		BatchMessageListener<Integer, String> batchMessageListener = spy(
 				new BatchMessageListener<Integer, String>() { // Cannot be lambda: Mockito doesn't mock final classes
 
 					@Override
 					public void onMessage(List<ConsumerRecord<Integer, String>> data) {
-						latch.countDown();
-						if (latch.getCount() > 0) {
+						if (attempts.getAndDecrement() > 0) {
 							throw new IllegalArgumentException("Failed record");
 						}
 					}
@@ -4334,6 +4334,8 @@ public void onMessage(List<ConsumerRecord<Integer, String>> data) {
 		containerProps.setMessageListener(batchMessageListener);
 		containerProps.setClientId("clientId");
 
+		CountDownLatch successLatch = new CountDownLatch(1);
+
 		BatchInterceptor<Integer, String> batchInterceptor = spy(new BatchInterceptor<Integer, String>() {
 
 			@Override
@@ -4342,14 +4344,20 @@ public ConsumerRecords<Integer, String> intercept(ConsumerRecords<Integer, Strin
 				return records;
 			}
 
+			@Override
+			public void success(ConsumerRecords<Integer, String> records, Consumer<Integer, String> consumer) {
+				successLatch.countDown();
+			}
+
 		});
 
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(0, 3)));
 		container.setBatchInterceptor(batchInterceptor);
 		container.start();
-		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+
+		assertThat(successLatch.await(10, TimeUnit.SECONDS)).isTrue();
 
 		InOrder inOrder = inOrder(batchInterceptor, batchMessageListener, consumer);
 		for (int i = 0; i < 3; i++) {