GH-1424: Abort delivery on ProducerFencedException

Resolves #1424

Do not deliver remaining records after a `ProducerFencedException` - the
partitions will have been reassigned to another instance.

**cherry-pick to 2.4.x, 2.3.x**

Author: garyrussell

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

@@ -1559,6 +1559,7 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 				}
 				catch (ProducerFencedException e) {
 					this.logger.error(e, "Producer fenced during transaction");
+					break;
 				}
 				catch (RuntimeException e) {
 					this.logger.error(e, "Transaction rolled back");

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

@@ -46,6 +46,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 
@@ -715,7 +716,9 @@ void testNoAfterRollbackWhenFenced() throws Exception {
 		recordMap.put(topicPartition1, Collections.singletonList(new ConsumerRecord<>("foo", 1, 0, "key", "value")));
 		ConsumerRecords records = new ConsumerRecords(recordMap);
 		final AtomicBoolean done = new AtomicBoolean();
+		final CountDownLatch pollLatch = new CountDownLatch(2);
 		willAnswer(i -> {
+			pollLatch.countDown();
 			if (done.compareAndSet(false, true)) {
 				return records;
 			}
@@ -727,11 +730,6 @@ void testNoAfterRollbackWhenFenced() throws Exception {
 		ConsumerFactory cf = mock(ConsumerFactory.class);
 		willReturn(consumer).given(cf).createConsumer("group", "", null, KafkaTestUtils.defaultPropertyOverrides());
 		Producer producer = mock(Producer.class);
-		final CountDownLatch closeLatch = new CountDownLatch(1);
-		willAnswer(i -> {
-			closeLatch.countDown();
-			return null;
-		}).given(producer).close(any());
 		willThrow(new ProducerFencedException("test")).given(producer).commitTransaction();
 		ProducerFactory pf = mock(ProducerFactory.class);
 		given(pf.transactionCapable()).willReturn(true);
@@ -741,19 +739,23 @@ void testNoAfterRollbackWhenFenced() throws Exception {
 				new TopicPartitionOffset("foo", 1));
 		props.setGroupId("group");
 		props.setTransactionManager(tm);
+		AtomicInteger deliveryCount = new AtomicInteger();
 		props.setMessageListener((MessageListener) m -> {
+			deliveryCount.incrementAndGet();
 		});
 		KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, props);
 		AfterRollbackProcessor arp = mock(AfterRollbackProcessor.class);
 		given(arp.isProcessInTransaction()).willReturn(true);
 		container.setAfterRollbackProcessor(arp);
 		container.setBeanName("rollback");
 		container.start();
-		assertThat(closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		InOrder inOrder = inOrder(producer);
 		inOrder.verify(producer).beginTransaction();
 		inOrder.verify(producer).commitTransaction();
 		inOrder.verify(producer).close(any());
+		inOrder.verifyNoMoreInteractions();
+		assertThat(deliveryCount.get()).isEqualTo(1);
 
 		verify(arp, never()).process(any(), any(), any(), anyBoolean());