GH-800: Fix Zombie Fencing

Resolves https://github.com/spring-projects/spring-kafka/issues/800

Fix assignment of `transactional.id` to be consistent across consumers.

**cherry-pick to all versions >= 1.3.x**

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
#	spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
#	spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java
#	src/reference/asciidoc/whats-new.adoc

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

@@ -18,13 +18,16 @@ package org.springframework.kafka.core;
 
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -44,6 +47,7 @@ import org.apache.kafka.common.serialization.Serializer;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.Lifecycle;
+import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.util.Assert;
 
 /**
@@ -81,6 +85,8 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 	private final BlockingQueue<CloseSafeProducer<K, V>> cache = new LinkedBlockingQueue<>();
 
+	private final Map<String, Producer<K, V>> consumerProducers = new HashMap<>();
+
 	private volatile CloseSafeProducer<K, V> producer;
 
 	private Serializer<K> keySerializer;
@@ -93,6 +99,12 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 	private volatile boolean running;
 
+	private boolean producerPerConsumerPartition = true;
+
+	/**
+	 * Construct a factory with the provided configuration.
+	 * @param configs the configuration.
+	 */
 	public DefaultKafkaProducerFactory(Map<String, Object> configs) {
 		this(configs, null, null);
 	}
@@ -144,6 +156,17 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 		}
 	}
 
+	/**
+	 * Set to false to revert to the previous behavior of a simple incrementing
+	 * trasactional.id suffix for each producer instead of maintaining a producer
+	 * for each group/topic/partition.
+	 * @param producerPerConsumerPartition false to revert.
+	 * @since 1.3.7
+	 */
+	public void setProducerPerConsumerPartition(boolean producerPerConsumerPartition) {
+		this.producerPerConsumerPartition = producerPerConsumerPartition;
+	}
+
 	/**
 	 * Return an unmodifiable reference to the configuration map for this factory.
 	 * Useful for cloning to make a similar factory.
@@ -177,6 +200,12 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 			}
 			producer = this.cache.poll();
 		}
+		synchronized (this.consumerProducers) {
+			this.consumerProducers.forEach(
+					(k, v) -> ((CloseSafeProducer<K, V>) v).delegate
+							.close(this.physicalCloseTimeout, TimeUnit.SECONDS));
+			this.consumerProducers.clear();
+		}
 	}
 
 	@Override
@@ -205,7 +234,12 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 	@Override
 	public Producer<K, V> createProducer() {
 		if (this.transactionIdPrefix != null) {
-			return createTransactionalProducer();
+			if (this.producerPerConsumerPartition) {
+				return createTransactionalProducerForPartition();
+			}
+			else {
+				return createTransactionalProducer();
+			}
 		}
 		if (this.producer == null) {
 			synchronized (this) {
@@ -226,6 +260,37 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 		return new KafkaProducer<K, V>(this.configs, this.keySerializer, this.valueSerializer);
 	}
 
+	private Producer<K, V> createTransactionalProducerForPartition() {
+		String suffix = TransactionSupport.getTransactionIdSuffix();
+		if (suffix == null) {
+			return createTransactionalProducer();
+		}
+		else {
+			synchronized (this.consumerProducers) {
+				if (!this.consumerProducers.containsKey(suffix)) {
+					Producer<K, V> newProducer = doCreateTxProducer(suffix, this::removeConsumerProducer);
+					this.consumerProducers.put(suffix, newProducer);
+					return newProducer;
+				}
+				else {
+					return this.consumerProducers.get(suffix);
+				}
+			}
+		}
+	}
+
+	private void removeConsumerProducer(CloseSafeProducer<K, V> producer) {
+		synchronized (this.consumerProducers) {
+			Iterator<Entry<String, Producer<K, V>>> iterator = this.consumerProducers.entrySet().iterator();
+			while (iterator.hasNext()) {
+				if (iterator.next().getValue().equals(producer)) {
+					iterator.remove();
+					break;
+				}
+			}
+		}
+	}
+
 	/**
 	 * Subclasses must return a producer from the {@link #getCache()} or a
 	 * new raw producer wrapped in a {@link CloseSafeProducer}.
@@ -235,18 +300,22 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 	protected Producer<K, V> createTransactionalProducer() {
 		Producer<K, V> producer = this.cache.poll();
 		if (producer == null) {
-			Map<String, Object> configs = new HashMap<>(this.configs);
-			configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
-					this.transactionIdPrefix + this.transactionIdSuffix.getAndIncrement());
-			producer = new KafkaProducer<K, V>(configs, this.keySerializer, this.valueSerializer);
-			producer.initTransactions();
-			return new CloseSafeProducer<K, V>(producer, this.cache);
+			return doCreateTxProducer("" + this.transactionIdSuffix.getAndIncrement(), null);
 		}
 		else {
 			return producer;
 		}
 	}
 
+	private Producer<K, V> doCreateTxProducer(String suffix, Consumer<CloseSafeProducer<K, V>> remover) {
+		Producer<K, V> producer;
+		Map<String, Object> configs = new HashMap<>(this.configs);
+		configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, this.transactionIdPrefix + suffix);
+		producer = new KafkaProducer<K, V>(configs, this.keySerializer, this.valueSerializer);
+		producer.initTransactions();
+		return new CloseSafeProducer<K, V>(producer, this.cache, remover);
+	}
+
 	protected BlockingQueue<CloseSafeProducer<K, V>> getCache() {
 		return this.cache;
 	}
@@ -264,16 +333,24 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 		private final BlockingQueue<CloseSafeProducer<K, V>> cache;
 
+		private final Consumer<CloseSafeProducer<K, V>> removeConsumerProducer;
+
 		private volatile boolean txFailed;
 
 		CloseSafeProducer(Producer<K, V> delegate) {
-			this(delegate, null);
+			this(delegate, null, null);
 			Assert.isTrue(!(delegate instanceof CloseSafeProducer), "Cannot double-wrap a producer");
 		}
 
 		CloseSafeProducer(Producer<K, V> delegate, BlockingQueue<CloseSafeProducer<K, V>> cache) {
+			this(delegate, cache, null);
+		}
+
+		CloseSafeProducer(Producer<K, V> delegate, BlockingQueue<CloseSafeProducer<K, V>> cache,
+				Consumer<CloseSafeProducer<K, V>> removeConsumerProducer) {
 			this.delegate = delegate;
 			this.cache = cache;
+			this.removeConsumerProducer = removeConsumerProducer;
 		}
 
 		@Override
@@ -353,6 +430,9 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 							+ "broker restarted during transaction");
 
 					this.delegate.close();
+					if (this.removeConsumerProducer != null) {
+						this.removeConsumerProducer.accept(this);
+					}
 				}
 				else {
 					synchronized (this) {

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

@@ -65,6 +65,7 @@ import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.LogIfLevelEnabled;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition;
+import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
 import org.springframework.scheduling.SchedulingAwareRunnable;
 import org.springframework.scheduling.TaskScheduler;
@@ -1012,6 +1013,8 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 					this.logger.trace("Processing " + record);
 				}
 				try {
+					TransactionSupport.setTransactionIdSuffix(
+							this.consumerGroupId + "." + record.topic() + "." + record.partition());
 					this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
 
 						@Override
@@ -1038,6 +1041,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 					}
 					getAfterRollbackProcessor().process(unprocessed, this.consumer);
 				}
+				finally {
+					TransactionSupport.clearTransactionIdSuffix();
+				}
 			}
 		}
 

spring-kafka/src/main/java/org/springframework/kafka/support/TransactionSupport.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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 org.springframework.kafka.support;
+
+/**
+ * Utilities for supporting transactions.
+ *
+ * @author Gary Russell
+ * @since 1.3.7
+ *
+ */
+public final class TransactionSupport {
+
+	private static final ThreadLocal<String> transactionIdSuffix = new ThreadLocal<>();
+
+	private TransactionSupport() {
+		super();
+	}
+
+	public static void setTransactionIdSuffix(String suffix) {
+		transactionIdSuffix.set(suffix);
+	}
+
+	public static String getTransactionIdSuffix() {
+		return transactionIdSuffix.get();
+	}
+
+	public static void clearTransactionIdSuffix() {
+		transactionIdSuffix.remove();
+	}
+
+}

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

@@ -39,6 +39,7 @@ import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -63,6 +64,7 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.core.ProducerFactoryUtils;
 import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
@@ -160,7 +162,8 @@ public class TransactionalContainerTests {
 			}
 		});
 		if (handleError) {
-			props.setErrorHandler((e, data) -> { });
+			props.setErrorHandler((e, data) -> {
+			});
 		}
 		KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, props);
 		container.setBeanName("commit");
@@ -377,6 +380,7 @@ public class TransactionalContainerTests {
 		verify(pf).createProducer();
 	}
 
+	@SuppressWarnings("unchecked")
 	@Test
 	public void testRollbackRecord() throws Exception {
 		logger.info("Start testRollbackRecord");
@@ -397,6 +401,7 @@ public class TransactionalContainerTests {
 		final KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		final AtomicBoolean failed = new AtomicBoolean();
 		final CountDownLatch latch = new CountDownLatch(3);
+		final AtomicReference<String> transactionalId = new AtomicReference<>();
 		containerProps.setMessageListener((MessageListener<Integer, String>) message -> {
 			latch.countDown();
 			if (failed.compareAndSet(false, true)) {
@@ -408,6 +413,9 @@ public class TransactionalContainerTests {
 			if (message.topic().equals(topic1)) {
 				template.send(topic2, "bar");
 				template.flush();
+				transactionalId.set(KafkaTestUtils.getPropertyValue(
+						ProducerFactoryUtils.getTransactionalResourceHolder(pf).getProducer(),
+						"delegate.transactionManager.transactionalId", String.class));
 			}
 		});
 
@@ -444,8 +452,11 @@ public class TransactionalContainerTests {
 		ConsumerRecords<Integer, String> records = consumer.poll(0);
 		assertThat(records.count()).isEqualTo(0);
 		assertThat(consumer.position(new TopicPartition(topic1, 0))).isEqualTo(1);
+		assertThat(transactionalId.get()).startsWith("rr.group.txTopic");
 		logger.info("Stop testRollbackRecord");
 		pf.destroy();
+		assertThat(KafkaTestUtils.getPropertyValue(pf, "consumerProducers", Map.class)).isEmpty();
+		consumer.close();
 	}
 
 	@SuppressWarnings("serial")

src/reference/asciidoc/kafka.adoc

@@ -256,7 +256,12 @@ Spring for Apache Kafka adds support in several ways.
 Transactions are enabled by providing the `DefaultKafkaProducerFactory` with a `transactionIdPrefix`.
 In that case, instead of managing a single shared `Producer`, the factory maintains a cache of transactional producers.
 When the user `close()` s a producer, it is returned to the cache for reuse instead of actually being closed.
-The `transactional.id` property of each producer is `transactionIdPrefix` + `n`, where `n` starts with `0` and is incremented for each new producer.
+The `transactional.id` property of each producer is `transactionIdPrefix` + `n`, where `n` starts with `0` and is incremented for each new producer, unless the transaction is started by a listener container with a record-based listener.
+In that case, the `transactional.id` is `<transactionIdPrefix>.<group.id>.<topic>.<partition>`; this is to properly support fencing zombies https://www.confluent.io/blog/transactions-apache-kafka/[as described here].
+This new behavior was added in versions 1.3.7, 2.0.6, 2.1.10, and 2.2.0.
+If you wish to revert to the previous behavior, set the `producerPerConsumerPartition` property on the `DefaultKafkaProducerFactory` to `false`.
+
+NOTE: While transactions are supported with batch listeners, zombie fencing cannot be supported because a batch may contain records from multiple topics/partitions.
 
 ====== KafkaTransactionManager
 

src/reference/asciidoc/whats-new.adoc

@@ -59,6 +59,11 @@ _Version 2.1.3_ introduced the `ChainedKafkaTransactionManager` see <<chained-tr
 Starting with _version 2.1.6_, a new `AfterRollbackProcessor` strategy is provided - see <<after-rollback>> for more information.
 
 
+==== Transactional Id
+
+When a transaction is started by the listener container, the `transactional.id` is now the `transactionIdPrefix` appended with `<group.id>.<topic>.<partition>`.
+This is to allow proper fencing of zombies https://www.confluent.io/blog/transactions-apache-kafka/[as described here].
+
 ==== Migration Guide from 2.0
 
 https://github.com/spring-projects/spring-kafka/wiki/Spring-for-Apache-Kafka-2.0-to-2.1-Migration-Guide[2.0 to 2.1 Migration].