GH-800: Fix Zombie Fencing

Resolves #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

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

Author: garyrussell

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

@@ -18,13 +18,16 @@
 
 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.springframework.beans.factory.DisposableBean;
 import org.springframework.context.Lifecycle;
+import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.util.Assert;
 
 /**
@@ -80,6 +84,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;
@@ -92,6 +98,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);
 	}
@@ -131,6 +143,17 @@ public void setTransactionIdPrefix(String transactionIdPrefix) {
 		this.transactionIdPrefix = transactionIdPrefix;
 	}
 
+	/**
+	 * 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.
@@ -164,6 +187,12 @@ public void destroy() throws Exception { //NOSONAR
 			}
 			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
@@ -192,7 +221,12 @@ public boolean isRunning() {
 	@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) {
@@ -213,6 +247,37 @@ protected Producer<K, V> createKafkaProducer() {
 		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}.
@@ -222,18 +287,22 @@ protected Producer<K, V> createKafkaProducer() {
 	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;
 	}
@@ -251,16 +320,24 @@ protected static class CloseSafeProducer<K, V> implements Producer<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
@@ -340,6 +417,9 @@ public void close() {
 							+ "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

@@ -59,6 +59,7 @@
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition;
+import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
 import org.springframework.scheduling.SchedulingAwareRunnable;
 import org.springframework.scheduling.TaskScheduler;
@@ -362,7 +363,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 		@SuppressWarnings("unchecked")
 		ListenerConsumer(GenericMessageListener<?> listener, ListenerType listenerType) {
 			Assert.state(!this.isAnyManualAck || !this.autoCommit,
-					"Consumer cannot be configured for auto commit for ackMode " + this.containerProperties.getAckMode());
+					"Consumer cannot be configured for auto commit for ackMode " + this.containerProperties
+							.getAckMode());
 			final Consumer<K, V> consumer = KafkaMessageListenerContainer.this.consumerFactory.createConsumer(
 					this.consumerGroupId, KafkaMessageListenerContainer.this.clientIdSuffix);
 			this.consumer = consumer;
@@ -789,7 +791,8 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 						Producer producer = null;
 						if (ListenerConsumer.this.kafkaTxManager != null) {
 							producer = ((KafkaResourceHolder) TransactionSynchronizationManager
-									.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory())).getProducer();
+									.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
+									.getProducer();
 						}
 						RuntimeException aborted = doInvokeBatchListener(records, recordList, producer);
 						if (aborted != null) {
@@ -901,14 +904,17 @@ private void innvokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
 					this.logger.trace("Processing " + record);
 				}
 				try {
+					TransactionSupport.setTransactionIdSuffix(
+							this.consumerGroupId + "." + record.topic() + "." + record.partition());
 					this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
 
 						@Override
 						public void doInTransactionWithoutResult(TransactionStatus s) {
 							Producer producer = null;
 							if (ListenerConsumer.this.kafkaTxManager != null) {
 								producer = ((KafkaResourceHolder) TransactionSynchronizationManager
-										.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory())).getProducer();
+										.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
+										.getProducer();
 							}
 							RuntimeException aborted = doInvokeRecordListener(record, producer, iterator);
 							if (aborted != null) {
@@ -927,6 +933,9 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 					}
 					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.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.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;
@@ -150,7 +152,8 @@ private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handl
 			}
 		});
 		if (handleError) {
-			props.setErrorHandler((e, data) -> { });
+			props.setErrorHandler((e, data) -> {
+			});
 		}
 		KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, props);
 		container.setBeanName("commit");
@@ -367,6 +370,7 @@ public void testConsumeAndProduceTransactionExternalTM() throws Exception {
 		verify(pf).createProducer();
 	}
 
+	@SuppressWarnings("unchecked")
 	@Test
 	public void testRollbackRecord() throws Exception {
 		logger.info("Start testRollbackRecord");
@@ -387,6 +391,7 @@ public void testRollbackRecord() throws Exception {
 		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)) {
@@ -398,6 +403,9 @@ public void testRollbackRecord() throws Exception {
 			if (message.topic().equals(topic1)) {
 				template.send(topic2, "bar");
 				template.flush();
+				transactionalId.set(KafkaTestUtils.getPropertyValue(
+						ProducerFactoryUtils.getTransactionalResourceHolder(pf).getProducer(),
+						"delegate.transactionManager.transactionalId", String.class));
 			}
 		});
 
@@ -434,8 +442,11 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 		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

@@ -243,7 +243,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

@@ -28,3 +28,9 @@ See <<testing>> for more information.
 
 Starting with _version 2.0.5_, 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].
+
+