GH-2604: Fix Transaction Synch Regression (#2605)

Resolves #2604

Prior to #2005 synchronized transactions were committed in `afterCompletion`.
This was incorrect because a failure to commit was not propagated to the caller.
However, if a higher order synchronization throws an exception in `afterCommit`,
the producer is closed (returned to the cache) while still in a transaction -
the transaction was not committed.

Restore the commit in `afterCompletion` and add a check in `KafkaResourceHolder`
to only commit once.

Add tests.

**cherry-pick to 2.9.x**

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaResourceHolder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2023 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.
@@ -38,6 +38,8 @@ public class KafkaResourceHolder<K, V> extends ResourceHolderSupport {
 
 	private final Duration closeTimeout;
 
+	private boolean committed;
+
 	/**
 	 * Construct an instance for the producer.
 	 * @param producer the producer.
@@ -55,7 +57,10 @@ public Producer<K, V> getProducer() {
 	}
 
 	public void commit() {
-		this.producer.commitTransaction();
+		if (!this.committed) {
+			this.producer.commitTransaction();
+			this.committed = true;
+		}
 	}
 
 	public void close() {

spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactoryUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -153,7 +153,10 @@ protected void processResourceAfterCommit(KafkaResourceHolder<K, V> resourceHold
 		@Override
 		public void afterCompletion(int status) {
 			try {
-				if (status != TransactionSynchronization.STATUS_COMMITTED) {
+				if (status == TransactionSynchronization.STATUS_COMMITTED) {
+					this.resourceHolder.commit();
+				}
+				else {
 					this.resourceHolder.rollback();
 				}
 			}

spring-kafka/src/test/java/transaction/TransactionSynchronizationTests.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2023 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package transaction;
+
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import org.apache.kafka.clients.producer.Producer;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.core.Ordered;
+import org.springframework.kafka.core.KafkaResourceHolder;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.core.ProducerFactoryUtils;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionException;
+import org.springframework.transaction.support.AbstractPlatformTransactionManager;
+import org.springframework.transaction.support.DefaultTransactionStatus;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.transaction.support.TransactionTemplate;
+
+/**
+ * @author Gary Russell
+ * @since 2.9.7
+ *
+ */
+public class TransactionSynchronizationTests {
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	void commitAfterAnotherSyncFails() {
+		Producer producer = mock(Producer.class);
+		ProducerFactory pf = mock(ProducerFactory.class);
+		given(pf.createProducer(any())).willReturn(producer);
+		assertThatExceptionOfType(RuntimeException.class)
+				.isThrownBy(() ->
+					new TransactionTemplate(new TM()).executeWithoutResult(status -> {
+						KafkaResourceHolder holder = ProducerFactoryUtils.getTransactionalResourceHolder(pf);
+						TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+
+							@Override
+							public void afterCommit() {
+								if (true) {
+									throw new RuntimeException("Test");
+								}
+							}
+
+							@Override
+							public int getOrder() {
+								return Ordered.HIGHEST_PRECEDENCE;
+							}
+
+						});
+					}))
+				.withMessage("Test");
+		verify(producer).beginTransaction();
+		verify(producer).commitTransaction();
+		verify(producer).close(any());
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	void onlyOnceCommit() {
+		Producer producer = mock(Producer.class);
+		ProducerFactory pf = mock(ProducerFactory.class);
+		given(pf.createProducer(any())).willReturn(producer);
+		new TransactionTemplate(new TM()).executeWithoutResult(status -> {
+			KafkaResourceHolder holder = ProducerFactoryUtils.getTransactionalResourceHolder(pf);
+		});
+		verify(producer).beginTransaction();
+		verify(producer).commitTransaction();
+		verify(producer).close(any());
+	}
+
+	static class TM extends AbstractPlatformTransactionManager {
+
+		@Override
+		protected Object doGetTransaction() throws TransactionException {
+			return new Object();
+		}
+
+		@Override
+		protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException {
+		}
+
+		@Override
+		protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
+		}
+
+		@Override
+		protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
+		}
+
+	}
+
+}
+