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.

**cherry-pick to 2.1.x, 2.0.x**

I will backport to 1.3.x (without lambdas etc) after review/merge.

* checkstyle

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

Author: garyrussell

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

@@ -85,7 +85,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;
 
@@ -202,8 +202,7 @@ public void destroy() throws Exception { //NOSONAR
 		}
 		synchronized (this.consumerProducers) {
 			this.consumerProducers.forEach(
-					(k, v) -> ((CloseSafeProducer<K, V>) v).delegate
-							.close(this.physicalCloseTimeout, TimeUnit.SECONDS));
+				(k, v) -> v.delegate.close(this.physicalCloseTimeout, TimeUnit.SECONDS));
 			this.consumerProducers.clear();
 		}
 	}
@@ -268,7 +267,7 @@ private Producer<K, V> createTransactionalProducerForPartition() {
 		else {
 			synchronized (this.consumerProducers) {
 				if (!this.consumerProducers.containsKey(suffix)) {
-					Producer<K, V> newProducer = doCreateTxProducer(suffix, this::removeConsumerProducer);
+					CloseSafeProducer<K, V> newProducer = doCreateTxProducer(suffix, this::removeConsumerProducer);
 					this.consumerProducers.put(suffix, newProducer);
 					return newProducer;
 				}
@@ -281,7 +280,7 @@ private Producer<K, V> createTransactionalProducerForPartition() {
 
 	private void removeConsumerProducer(CloseSafeProducer<K, V> producer) {
 		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(producer)) {
 					iterator.remove();
@@ -307,7 +306,7 @@ protected Producer<K, V> createTransactionalProducer() {
 		}
 	}
 
-	private Producer<K, V> doCreateTxProducer(String suffix, Consumer<CloseSafeProducer<K, V>> remover) {
+	private CloseSafeProducer<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);
@@ -320,6 +319,18 @@ protected BlockingQueue<CloseSafeProducer<K, V>> getCache() {
 		return this.cache;
 	}
 
+	@Override
+	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.
 	 *

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

@@ -28,10 +28,27 @@
  */
 public interface ProducerFactory<K, V> {
 
+	/**
+	 * Create a producer.
+	 * @return the producer.
+	 */
 	Producer<K, V> createProducer();
 
+	/**
+	 * Return true if the factory supports transactions.
+	 * @return true if transactional.
+	 */
 	default boolean transactionCapable() {
 		return false;
 	}
 
+	/**
+	 * Remove the specified producer from the cache and close it.
+	 * @param transactionIdSuffix the producer's transaction id suffix.
+	 * @since 1.3.8
+	 */
+	default void closeProducerFor(String transactionIdSuffix) {
+		// NOSONAR
+	}
+
 }

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

@@ -54,6 +54,7 @@
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaResourceHolder;
+import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.core.ProducerFactoryUtils;
 import org.springframework.kafka.event.ConsumerPausedEvent;
 import org.springframework.kafka.event.ConsumerResumedEvent;
@@ -544,6 +545,9 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 					if (this.consumerAwareListener != null) {
 						this.consumerAwareListener.onPartitionsRevokedAfterCommit(consumer, partitions);
 					}
+					if (ListenerConsumer.this.kafkaTxManager != null) {
+						closeProducers(partitions);
+					}
 				}
 
 				@Override
@@ -768,6 +772,9 @@ public void run() {
 						// No-op. Continue process
 					}
 				}
+				else {
+					closeProducers(getAssignedPartitions());
+				}
 			}
 			else {
 				ListenerConsumer.this.logger.error("No offset and no reset policy; stopping container");
@@ -1019,8 +1026,8 @@ private void invokeRecordListenerInTx(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
@@ -1395,6 +1402,23 @@ public String toString() {
 					+ "]";
 		}
 
+		private void closeProducers(Collection<TopicPartition> partitions) {
+			ProducerFactory<?, ?> producerFactory = this.kafkaTxManager.getProducerFactory();
+			partitions.forEach(tp -> {
+				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

@@ -453,9 +453,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();
 	}