GH-800: Fix Zombie Fencing

Resolves #800

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

Backported to 1.3.x, removing Java 8 constructs.

Author: garyrussell

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

@@ -18,8 +18,10 @@
 
 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;
@@ -44,6 +46,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 +83,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 +97,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 +142,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 +186,13 @@ public void destroy() throws Exception { //NOSONAR
 			}
 			producer = this.cache.poll();
 		}
+		synchronized (this.consumerProducers) {
+			for (Entry<String, Producer<K, V>> entry : this.consumerProducers.entrySet()) {
+				((CloseSafeProducer<K, V>) entry.getValue()).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,25 @@ 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.consumerProducers.put(suffix, newProducer);
+					return newProducer;
+				}
+				else {
+					return this.consumerProducers.get(suffix);
+				}
+			}
+		}
+	}
+
 	/**
 	 * Subclasses must return a producer from the {@link #getCache()} or a
 	 * new raw producer wrapped in a {@link CloseSafeProducer}.
@@ -222,18 +275,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());
 		}
 		else {
 			return producer;
 		}
 	}
 
+	private Producer<K, V> doCreateTxProducer(String suffix) {
+		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, this.consumerProducers);
+	}
+
 	protected BlockingQueue<CloseSafeProducer<K, V>> getCache() {
 		return this.cache;
 	}
@@ -251,16 +308,24 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 
 		private final BlockingQueue<CloseSafeProducer<K, V>> cache;
 
+		private final Map<String, Producer<K, V>> consumerProducers;
+
 		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,
+				Map<String, Producer<K, V>> consumerProducers) {
 			this.delegate = delegate;
 			this.cache = cache;
+			this.consumerProducers = consumerProducers;
 		}
 
 		@Override
@@ -340,6 +405,9 @@ public void close() {
 							+ "broker restarted during transaction");
 
 					this.delegate.close();
+					if (this.consumerProducers != null) {
+						removeConsumerProducer();
+					}
 				}
 				else {
 					synchronized (this) {
@@ -356,6 +424,18 @@ public void close(long timeout, TimeUnit unit) {
 			close();
 		}
 
+		private void removeConsumerProducer() {
+			synchronized (this.consumerProducers) {
+				Iterator<Entry<String, Producer<K, V>>> iterator = this.consumerProducers.entrySet().iterator();
+				while (iterator.hasNext()) {
+					if (iterator.next().getValue().equals(this)) {
+						iterator.remove();
+						break;
+					}
+				}
+			}
+		}
+
 		@Override
 		public String toString() {
 			return "CloseSafeProducer [delegate=" + this.delegate + "]";

Diff for: 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;
@@ -894,6 +895,8 @@ 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
@@ -920,6 +923,9 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 					}
 					getAfterRollbackProcessor().process(unprocessed, this.consumer);
 				}
+				finally {
+					TransactionSupport.clearTransactionIdSuffix();
+				}
 			}
 		}
 

Diff for: 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();
+	}
+
+}

Diff for: 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")

Diff for: src/reference/asciidoc/kafka.adoc

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

Diff for: src/reference/asciidoc/whats-new.adoc

@@ -46,3 +46,8 @@ Starting with _version 1.3.5_, a new `AfterRollbackProcessor` strategy is provid
 
 Support for configuring Kerberos is now provided.
 See <<kerberos>> 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].