GH-859: Fix nested transactions

Resolves #859

When using `executeInTransaction` on a transactional container thread,
we cannot use the existing transaction - clear the TL to allow a new
Producer to be allocated.

Invalid state transition (in-trans to in-trans).

**cherry-pick to 2.1.x, 2.0.x, backport needed for 1.3.x**

# Conflicts:
#	spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java

Author: garyrussell

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

@@ -154,6 +154,16 @@ public void setProducerPerConsumerPartition(boolean producerPerConsumerPartition
 		this.producerPerConsumerPartition = producerPerConsumerPartition;
 	}
 
+	/**
+	 * Return the producerPerConsumerPartition.
+	 * @return the producerPerConsumerPartition.
+	 * @since 1.3.8
+	 */
+	@Override
+	public boolean isProducerPerConsumerPartition() {
+		return this.producerPerConsumerPartition;
+	}
+
 	/**
 	 * Return an unmodifiable reference to the configuration map for this factory.
 	 * Useful for cloning to make a similar factory.
@@ -246,7 +256,7 @@ protected Producer<K, V> createKafkaProducer() {
 		return new KafkaProducer<K, V>(this.configs, this.keySerializer, this.valueSerializer);
 	}
 
-	private Producer<K, V> createTransactionalProducerForPartition() {
+	Producer<K, V> createTransactionalProducerForPartition() {
 		String suffix = TransactionSupport.getTransactionIdSuffix();
 		if (suffix == null) {
 			return createTransactionalProducer();

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

@@ -34,6 +34,7 @@
 import org.springframework.kafka.support.LoggingProducerListener;
 import org.springframework.kafka.support.ProducerListener;
 import org.springframework.kafka.support.SendResult;
+import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
@@ -236,6 +237,15 @@ public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
 		Assert.state(this.transactional, "Producer factory does not support transactions");
 		Producer<K, V> producer = this.producers.get();
 		Assert.state(producer == null, "Nested calls to 'executeInTransaction' are not allowed");
+		String transactionIdSuffix;
+		if (this.producerFactory.isProducerPerConsumerPartition()) {
+			transactionIdSuffix = TransactionSupport.getTransactionIdSuffix();
+			TransactionSupport.clearTransactionIdSuffix();
+		}
+		else {
+			transactionIdSuffix = null;
+		}
+
 		producer = this.producerFactory.createProducer();
 
 		try {
@@ -265,6 +275,9 @@ public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
 			producer.commitTransaction();
 		}
 		finally {
+			if (transactionIdSuffix != null) {
+				TransactionSupport.setTransactionIdSuffix(transactionIdSuffix);
+			}
 			this.producers.remove();
 			closeProducer(producer, false);
 		}

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

@@ -51,4 +51,13 @@ default void closeProducerFor(String transactionIdSuffix) {
 		// NOSONAR
 	}
 
+	/**
+	 * Return the producerPerConsumerPartition.
+	 * @return the producerPerConsumerPartition.
+	 * @since 1.3.8
+	 */
+	default boolean isProducerPerConsumerPartition() {
+		return false;
+	}
+
 }

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java

@@ -28,9 +28,11 @@
 import static org.springframework.kafka.test.assertj.KafkaConditions.key;
 import static org.springframework.kafka.test.assertj.KafkaConditions.value;
 
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -51,6 +53,7 @@
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.kafka.support.transaction.ResourcelessTransactionManager;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -234,6 +237,58 @@ public void testTransactionSynchronizationExceptionOnCommit() {
 		assertThat(producer.closed()).isTrue();
 	}
 
+	@Test
+	public void testExcecuteInTransactionNewInnerTx() {
+		@SuppressWarnings("unchecked")
+		Producer<Object, Object> producer1 = mock(Producer.class);
+		@SuppressWarnings("unchecked")
+		Producer<Object, Object> producer2 = mock(Producer.class);
+		producer1.initTransactions();
+		AtomicBoolean first = new AtomicBoolean(true);
+
+		DefaultKafkaProducerFactory<Object, Object> pf = new DefaultKafkaProducerFactory<Object, Object>(
+				Collections.emptyMap()) {
+
+			@Override
+			protected Producer<Object, Object> createTransactionalProducer() {
+				return first.getAndSet(false) ? producer1 : producer2;
+			}
+
+			@Override
+			Producer<Object, Object> createTransactionalProducerForPartition() {
+				return createTransactionalProducer();
+			}
+
+		};
+		pf.setTransactionIdPrefix("tx.");
+
+		KafkaTemplate<Object, Object> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(STRING_KEY_TOPIC);
+
+		KafkaTransactionManager<Object, Object> tm = new KafkaTransactionManager<>(pf);
+
+		try {
+			TransactionSupport.setTransactionIdSuffix("testExcecuteInTransactionNewInnerTx");
+			new TransactionTemplate(tm).execute(s -> {
+				return template.executeInTransaction(t -> {
+					template.sendDefault("foo", "bar");
+					return null;
+				});
+			});
+
+			InOrder inOrder = inOrder(producer1, producer2);
+			inOrder.verify(producer1).beginTransaction();
+			inOrder.verify(producer2).beginTransaction();
+			inOrder.verify(producer2).commitTransaction();
+			inOrder.verify(producer2).close();
+			inOrder.verify(producer1).commitTransaction();
+			inOrder.verify(producer1).close();
+		}
+		finally {
+			TransactionSupport.clearTransactionIdSuffix();
+		}
+	}
+
 	@Configuration
 	@EnableTransactionManagement
 	public static class DeclarativeConfig {