GH-834: Remove transactional producers

Resolves #834

To solve the zombie fencing problem there is a producer for each
group/topic/partition.

Close these producers when a partition is revoked or the container stopped.

Author: garyrussell

.gitignore

@@ -11,3 +11,4 @@ bin
 build
 out
 target
+.DS_Store

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

@@ -83,7 +83,7 @@ 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 final Map<String, CloseSafeProducer<K, V>> consumerProducers = new HashMap<>();
 
 	private volatile CloseSafeProducer<K, V> producer;
 
@@ -187,8 +187,8 @@ 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
+			for (Entry<String, CloseSafeProducer<K, V>> entry : this.consumerProducers.entrySet()) {
+				entry.getValue().delegate
 							.close(this.physicalCloseTimeout, TimeUnit.SECONDS);
 			}
 			this.consumerProducers.clear();
@@ -255,7 +255,7 @@ private Producer<K, V> createTransactionalProducerForPartition() {
 		else {
 			synchronized (this.consumerProducers) {
 				if (!this.consumerProducers.containsKey(suffix)) {
-					Producer<K, V> newProducer = doCreateTxProducer(suffix);
+					CloseSafeProducer<K, V> newProducer = doCreateTxProducer(suffix);
 					this.consumerProducers.put(suffix, newProducer);
 					return newProducer;
 				}
@@ -282,7 +282,7 @@ protected Producer<K, V> createTransactionalProducer() {
 		}
 	}
 
-	private Producer<K, V> doCreateTxProducer(String suffix) {
+	private CloseSafeProducer<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);
@@ -295,6 +295,17 @@ protected BlockingQueue<CloseSafeProducer<K, V>> getCache() {
 		return this.cache;
 	}
 
+	public void closeProducerFor(String transactionIdSuffix) {
+		if (this.producerPerConsumerPartition) {
+			synchronized (this.consumerProducers) {
+				CloseSafeProducer<K, V> removed = this.consumerProducers.remove(transactionIdSuffix);
+				if (removed != null) {
+					removed.delegate.close(this.physicalCloseTimeout, TimeUnit.SECONDS);
+				}
+			}
+		}
+	}
+
 	/**
 	 * A wrapper class for the delegate.
 	 *
@@ -308,7 +319,7 @@ 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 final Map<String, CloseSafeProducer<K, V>> consumerProducers;
 
 		private volatile boolean txFailed;
 
@@ -322,7 +333,7 @@ protected static class CloseSafeProducer<K, V> implements Producer<K, V> {
 		}
 
 		CloseSafeProducer(Producer<K, V> delegate, BlockingQueue<CloseSafeProducer<K, V>> cache,
-				Map<String, Producer<K, V>> consumerProducers) {
+				Map<String, CloseSafeProducer<K, V>> consumerProducers) {
 			this.delegate = delegate;
 			this.cache = cache;
 			this.consumerProducers = consumerProducers;
@@ -426,7 +437,8 @@ public void close(long timeout, TimeUnit unit) {
 
 		private void removeConsumerProducer() {
 			synchronized (this.consumerProducers) {
-				Iterator<Entry<String, Producer<K, V>>> iterator = this.consumerProducers.entrySet().iterator();
+				Iterator<Entry<String, CloseSafeProducer<K, V>>> iterator = this.consumerProducers.entrySet()
+						.iterator();
 				while (iterator.hasNext()) {
 					if (iterator.next().getValue().equals(this)) {
 						iterator.remove();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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.
@@ -50,6 +50,7 @@
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaResourceHolder;
 import org.springframework.kafka.core.ProducerFactoryUtils;
 import org.springframework.kafka.event.ListenerContainerIdleEvent;
@@ -489,6 +490,9 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 					getContainerProperties().getConsumerRebalanceListener().onPartitionsRevoked(partitions);
 					// Wait until now to commit, in case the user listener added acks
 					commitPendingAcks();
+					if (ListenerConsumer.this.kafkaTxManager != null) {
+						closeProducers(partitions);
+					}
 				}
 
 				@Override
@@ -680,6 +684,9 @@ public void run() {
 						// No-op. Continue process
 					}
 				}
+				else {
+					closeProducers(getAssignedPartitions());
+				}
 			}
 			else {
 				ListenerConsumer.this.logger.error("No offset and no reset policy; stopping container");
@@ -895,8 +902,8 @@ private void innvokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
 					this.logger.trace("Processing " + record);
 				}
 				try {
-					TransactionSupport.setTransactionIdSuffix(
-							this.consumerGroupId + "." + record.topic() + "." + record.partition());
+					TransactionSupport
+							.setTransactionIdSuffix(zombieFenceTxIdSuffix(record.topic(), record.partition()));
 					this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
 
 						@Override
@@ -1228,6 +1235,26 @@ public void seekToEnd(String topic, int partition) {
 			this.seeks.add(new TopicPartitionInitialOffset(topic, partition, SeekPosition.END));
 		}
 
+		private void closeProducers(Collection<TopicPartition> partitions) {
+			if (this.kafkaTxManager.getProducerFactory() instanceof DefaultKafkaProducerFactory) {
+				DefaultKafkaProducerFactory<?, ?> producerFactory =
+						(DefaultKafkaProducerFactory<?, ?>) this.kafkaTxManager.getProducerFactory();
+				for (TopicPartition tp : partitions) {
+					try {
+						producerFactory.closeProducerFor(zombieFenceTxIdSuffix(tp.topic(), tp.partition()));
+					}
+					catch (Exception e) {
+						this.logger.error("Failed to close producer with transaction id suffix: "
+								+ zombieFenceTxIdSuffix(tp.topic(), tp.partition()), e);
+					}
+				}
+			}
+		}
+
+		private String zombieFenceTxIdSuffix(String topic, int partition) {
+			return this.consumerGroupId + "." + topic + "." + partition;
+		}
+
 		private final class ConsumerAcknowledgment implements Acknowledgment {
 
 			private final ConsumerRecord<K, V> record;

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

@@ -443,9 +443,9 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 		assertThat(records.count()).isEqualTo(0);
 		assertThat(consumer.position(new TopicPartition(topic1, 0))).isEqualTo(1);
 		assertThat(transactionalId.get()).startsWith("rr.group.txTopic");
+		assertThat(KafkaTestUtils.getPropertyValue(pf, "consumerProducers", Map.class)).isEmpty();
 		logger.info("Stop testRollbackRecord");
 		pf.destroy();
-		assertThat(KafkaTestUtils.getPropertyValue(pf, "consumerProducers", Map.class)).isEmpty();
 		consumer.close();
 	}