GH-8: Shared KafkaProducer

Resolves: #8

Polishing - PR Comments

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

@@ -16,26 +16,128 @@
 package org.springframework.kafka.core;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.PartitionInfo;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.context.Lifecycle;
 
 /**
  * @author Gary Russell
  *
  */
-public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V> {
+public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>, Lifecycle, DisposableBean {
+
+	private static final Log logger = LogFactory.getLog(DefaultKafkaProducerFactory.class);
 
 	private final Map<String, Object> configs;
 
+	private volatile CloseSafeProducer<K, V> producer;
+
+	private volatile boolean running;
+
 	public DefaultKafkaProducerFactory(Map<String, Object> configs) {
 		this.configs = new HashMap<>(configs);
 	}
 
+	@Override
+	public void destroy() throws Exception {
+		CloseSafeProducer<K, V> producer = this.producer;
+		this.producer = null;
+		if (producer != null) {
+			producer.delegate.close();
+		}
+	}
+
+
+	@Override
+	public void start() {
+		this.running = true;
+	}
+
+
+	@Override
+	public void stop() {
+		try {
+			destroy();
+		}
+		catch (Exception e) {
+			logger.error("Exception while stopping producer", e);
+		}
+	}
+
+
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
 	@Override
 	public Producer<K, V> createProducer() {
-		return new KafkaProducer<>(this.configs);
+		if (this.producer == null) {
+			synchronized (this) {
+				if (this.producer == null) {
+					this.producer = new CloseSafeProducer<K, V>(new KafkaProducer<K, V>(this.configs));
+				}
+			}
+		}
+		return this.producer;
+	}
+
+	private static class CloseSafeProducer<K, V> implements Producer<K, V> {
+
+		private final Producer<K, V> delegate;
+
+		public CloseSafeProducer(Producer<K, V> delegate) {
+			this.delegate = delegate;
+		}
+
+		@Override
+		public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
+			return this.delegate.send(record);
+		}
+
+		@Override
+		public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
+			return this.delegate.send(record, callback);
+		}
+
+		@Override
+		public void flush() {
+			this.delegate.flush();
+		}
+
+		@Override
+		public List<PartitionInfo> partitionsFor(String topic) {
+			return this.delegate.partitionsFor(topic);
+		}
+
+		@Override
+		public Map<MetricName, ? extends Metric> metrics() {
+			return this.delegate.metrics();
+		}
+
+		@Override
+		public void close() {
+		}
+
+		@Override
+		public void close(long timeout, TimeUnit unit) {
+		}
+
 	}
 
 }