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

Author: garyrussell

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

@@ -95,7 +95,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;
 
@@ -224,7 +224,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();
 		}
 	}
@@ -316,7 +316,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;
 				}
@@ -329,7 +329,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();
@@ -355,7 +355,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);
@@ -368,6 +368,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

@@ -55,6 +55,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;
@@ -562,6 +563,9 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 					if (this.consumerAwareListener != null) {
 						this.consumerAwareListener.onPartitionsRevokedAfterCommit(consumer, partitions);
 					}
+					if (ListenerConsumer.this.kafkaTxManager != null) {
+						closeProducers(partitions);
+					}
 				}
 
 				@Override
@@ -786,6 +790,9 @@ public void run() {
 						// No-op. Continue process
 					}
 				}
+				else {
+					closeProducers(getAssignedPartitions());
+				}
 			}
 			else {
 				ListenerConsumer.this.logger.error("No offset and no reset policy; stopping container");
@@ -1064,8 +1071,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
@@ -1447,6 +1454,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

@@ -473,9 +473,9 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 		// depending on timing, the position might include the offset representing the commit in the log
 		assertThat(consumer.position(new TopicPartition(topic1, 0))).isGreaterThanOrEqualTo(1L);
 		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();
 	}