Merge ProducerSupplier with DeliveryStrategy

Author: Gérald QUINTANA

src/example/resources/logback.xml

@@ -24,7 +24,7 @@
         <keyingStrategy class="com.github.danielwegener.logback.kafka.keying.NoKeyKeyingStrategy" />
 
         <!-- use async delivery. the application threads are not blocked by logging -->
-        <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" />
+        <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.LateAsyncDeliveryStrategy" />
 
         <!-- each <producerConfig> translates to regular kafka-client config (format: key=value) -->
         <!-- producer configs are documented here: https://kafka.apache.org/documentation.html#newproducerconfigs -->

src/main/java/com/github/danielwegener/logback/kafka/KafkaAppender.java

@@ -4,7 +4,7 @@
 import ch.qos.logback.core.Appender;
 import ch.qos.logback.core.spi.AppenderAttachableImpl;
 import com.github.danielwegener.logback.kafka.delivery.FailedDeliveryCallback;
-import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
@@ -15,7 +15,7 @@
 /**
  * @since 0.0.1
  */
-public class KafkaAppender<E> extends KafkaAppenderConfig<E> {
+public class KafkaAppender<E> extends KafkaAppenderConfig<E, byte[], byte[]> {
 
     /**
      * Kafka clients uses this prefix for its slf4j logging.
@@ -57,15 +57,15 @@ public void start() {
             partition = null;
         }
 
-        producerSupplier.start(this, producerConfig);
+        deliveryStrategy.start(producerConfig, failedDeliveryCallback);
 
         super.start();
     }
 
     @Override
     public void stop() {
         super.stop();
-        producerSupplier.stop();
+        deliveryStrategy.stop();
     }
 
     @Override
@@ -112,12 +112,7 @@ protected void append(E e) {
 
         final ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(topic, partition, timestamp, key, payload);
 
-        final Producer<byte[], byte[]> producer = producerSupplier.get();
-        if (producer != null) {
-            deliveryStrategy.send(producer, record, e, failedDeliveryCallback);
-        } else {
-            failedDeliveryCallback.onFailedDelivery(e, null);
-        }
+        deliveryStrategy.send(record, e);
     }
 
     protected Long getTimestamp(E e) {

src/main/java/com/github/danielwegener/logback/kafka/KafkaAppenderConfig.java

@@ -3,35 +3,32 @@
 import ch.qos.logback.core.UnsynchronizedAppenderBase;
 import ch.qos.logback.core.encoder.Encoder;
 import ch.qos.logback.core.spi.AppenderAttachable;
-import com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy;
 import com.github.danielwegener.logback.kafka.delivery.DeliveryStrategy;
+import com.github.danielwegener.logback.kafka.delivery.LateAsyncDeliveryStrategy;
 import com.github.danielwegener.logback.kafka.keying.KeyingStrategy;
 import com.github.danielwegener.logback.kafka.keying.NoKeyKeyingStrategy;
-import com.github.danielwegener.logback.kafka.producer.LazyProducerSupplier;
-import com.github.danielwegener.logback.kafka.producer.ProducerSupplier;
 
 import java.util.HashMap;
 import java.util.Map;
 
-import static org.apache.kafka.clients.producer.ProducerConfig.*;
+import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG;
 
 /**
  * @since 0.0.1
  */
-public abstract class KafkaAppenderConfig<E> extends UnsynchronizedAppenderBase<E> implements AppenderAttachable<E> {
+public abstract class KafkaAppenderConfig<E, K, V> extends UnsynchronizedAppenderBase<E> implements AppenderAttachable<E> {
 
     protected String topic = null;
 
     protected Encoder<E> encoder = null;
     protected KeyingStrategy<? super E> keyingStrategy = null;
-    protected DeliveryStrategy deliveryStrategy;
+    protected DeliveryStrategy<E, K, V> deliveryStrategy;
 
     protected Integer partition = null;
 
     protected boolean appendTimestamp = true;
 
     protected Map<String,Object> producerConfig = new HashMap<String, Object>();
-    protected ProducerSupplier<byte[], byte[]> producerSupplier = new LazyProducerSupplier();
 
     protected boolean checkPrerequisites() {
         boolean errorFree = true;
@@ -58,8 +55,8 @@ protected boolean checkPrerequisites() {
         }
 
         if (deliveryStrategy == null) {
-            addInfo("No explicit deliveryStrategy set for the appender named [\""+name+"\"]. Using default asynchronous strategy.");
-            deliveryStrategy = new AsynchronousDeliveryStrategy();
+            addInfo("No explicit deliveryStrategy set for the appender named [\""+name+"\"]. Using default asynchronous lazy strategy.");
+            deliveryStrategy = new LateAsyncDeliveryStrategy();
         }
 
         return errorFree;
@@ -107,7 +104,4 @@ public void setAppendTimestamp(boolean appendTimestamp) {
         this.appendTimestamp = appendTimestamp;
     }
 
-    public void setProducerSupplier(ProducerSupplier<byte[], byte[]> producerSupplier) {
-        this.producerSupplier = producerSupplier;
-    }
 }

src/main/java/com/github/danielwegener/logback/kafka/delivery/AsynchronousDeliveryStrategy.java

@@ -0,0 +1,82 @@
+package com.github.danielwegener.logback.kafka.delivery;
+
+import ch.qos.logback.core.spi.ContextAwareBase;
+import org.apache.kafka.clients.producer.BufferExhaustedException;
+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.KafkaException;
+import org.apache.kafka.common.errors.TimeoutException;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Future;
+
+/**
+ * Base implementation of {@link DeliveryStrategy}
+ * @since 0.0.1
+ */
+public abstract class BaseDeliveryStrategy<E, K, V> extends ContextAwareBase implements DeliveryStrategy<E, K, V> {
+    private HashMap<String, Object> producerConfig;
+    protected FailedDeliveryCallback<E> failedDeliveryCallback;
+
+    @Override
+    public void start(Map<String, Object> producerConfig, FailedDeliveryCallback<E> failedDeliveryCallback) {
+        this.producerConfig = new HashMap<>(producerConfig);
+        this.failedDeliveryCallback = failedDeliveryCallback;
+    }
+
+    /**
+     * Instantiates the producer with its config
+     * @return Created producer
+     * @throws KafkaException Producer initialization failed
+     */
+    protected Producer<K, V> createProducer() throws KafkaException{
+        return new KafkaProducer<>(new HashMap<>(producerConfig));
+    }
+
+    /**
+     * Get producer
+     * @return Producer or null if producer is not initiliazed
+     */
+    protected abstract Producer<K, V> getProducer();
+    /**
+     * Stop the producer
+     */
+    public void stopProducer() {
+        if (getProducer() != null) {
+            try {
+                getProducer().close();
+            } catch (KafkaException e) {
+                addWarn("Failed to shut down kafka producer: " + e.getMessage(), e);
+            }
+        }
+    }
+
+    /**
+     * Send a record using the provider.
+     * In case of failure, the event is sent to the {@link #failedDeliveryCallback}
+     * @param record Record to send
+     * @param event Event to send to failedDeliveryCallback in case of failure
+     * @return Future returned by the Kafka producer
+     * @throws KafkaException Record sending failed
+     */
+    protected Future<RecordMetadata> doSend(ProducerRecord<K, V> record, final E event) throws KafkaException {
+        try {
+            return getProducer().send(record, new Callback() {
+                @Override
+                public void onCompletion(RecordMetadata metadata, Exception exception) {
+                    if (exception != null) {
+                        failedDeliveryCallback.onFailedDelivery(event, exception);
+                    }
+                }
+            });
+        } catch (BufferExhaustedException | TimeoutException e) {
+            failedDeliveryCallback.onFailedDelivery(event, e);
+            return null;
+        }
+    }
+
+}

src/main/java/com/github/danielwegener/logback/kafka/delivery/BaseDeliveryStrategy.java

@@ -1,11 +1,10 @@
 package com.github.danielwegener.logback.kafka.delivery;
 
-import ch.qos.logback.core.spi.ContextAwareBase;
-import org.apache.kafka.clients.producer.BufferExhaustedException;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 
+import java.util.Map;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -16,27 +15,45 @@
  * DeliveryStrategy that waits on the producer if the output buffer is full.
  * The wait timeout is configurable with {@link BlockingDeliveryStrategy#setTimeout(long)}
  * @since 0.0.1
- * @deprecated Use {@link AsynchronousDeliveryStrategy} instead.
+ * @deprecated Use {@link BaseDeliveryStrategy} instead.
  */
 @Deprecated
-public class BlockingDeliveryStrategy extends ContextAwareBase implements DeliveryStrategy {
+public class BlockingDeliveryStrategy<E, K, V> extends BaseDeliveryStrategy<E, K, V> {
 
     private long timeout = 0L;
+    private Producer<K,V> producer;
 
     @Override
-    public <K, V, E> boolean send(Producer<K, V> producer, ProducerRecord<K, V> record, E event, FailedDeliveryCallback<E> failureCallback) {
-        try {
+    public void start(Map<String, Object> producerConfig, FailedDeliveryCallback<E> failedDeliveryCallback) {
+        super.start(producerConfig, failedDeliveryCallback);
+        producer = createProducer();
+    }
 
-            final Future<RecordMetadata> future = producer.send(record);
-            if (timeout > 0L) future.get(timeout, TimeUnit.MILLISECONDS);
-            else if (timeout == 0) future.get();
-            return true;
+    @Override
+    public void send(ProducerRecord<K, V> record, E event) {
+        try {
+            final Future<RecordMetadata> future = doSend(record, event);
+            if (future == null) {
+                return;
+            }
+                if (timeout > 0L) future.get(timeout, TimeUnit.MILLISECONDS);
+                else if (timeout == 0) future.get();
         }
-        catch (InterruptedException e) { return false; }
-        catch (BufferExhaustedException | ExecutionException | CancellationException | TimeoutException e) {
-            failureCallback.onFailedDelivery(event, e);
+        catch (InterruptedException e) { }
+        catch (ExecutionException | CancellationException | TimeoutException e) {
+            failedDeliveryCallback.onFailedDelivery(event, e);
         }
-        return false;
+    }
+
+    @Override
+    public Producer<K, V> getProducer() {
+        return producer;
+    }
+
+    @Override
+    public void stop() {
+        stopProducer();
+        producer = null;
     }
 
     public long getTimeout() {

src/main/java/com/github/danielwegener/logback/kafka/delivery/BlockingDeliveryStrategy.java

@@ -1,26 +1,37 @@
 package com.github.danielwegener.logback.kafka.delivery;
 
+import ch.qos.logback.core.spi.ContextAware;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 
+import java.util.Map;
+
 /**
  * Interface for DeliveryStrategies.
+ *
+ * @param <E> the type of the logging event.
+ * @param <K> the key type of a persisted log message.
+ * @param <V> the value type of a persisted log message.
  * @since 0.0.1
  */
-public interface DeliveryStrategy {
+public interface DeliveryStrategy<E, K, V> {
+    /**
+     * Initialize the strategy and create Kafka producer if needed
+     * @param producerConfig Kafka producer config map
+     * @param failedDeliveryCallback a callback that handles messages that could not be delivered with best-effort.
+     */
+    void start(Map<String, Object> producerConfig, FailedDeliveryCallback<E> failedDeliveryCallback);
 
     /**
      * Sends a message to a kafka producer and somehow deals with failures.
      *
-     * @param producer the backing kafka producer
-     * @param record the prepared kafka message (ready to ship)
-     * @param event the originating logging event
-     * @param failedDeliveryCallback a callback that handles messages that could not be delivered with best-effort.
-     * @param <K> the key type of a persisted log message.
-     * @param <V> the value type of a persisted log message.
-     * @param <E> the type of the logging event.
-     * @return {@code true} if the message could be sent successfully, {@code false} otherwise.
+     * @param record   the prepared kafka message (ready to ship)
+     * @param event    the originating logging event
      */
-    <K,V,E> boolean send(Producer<K,V> producer, ProducerRecord<K, V> record, E event, FailedDeliveryCallback<E> failedDeliveryCallback);
+    void send(ProducerRecord<K, V> record, E event);
 
+    /**
+     * Stop this strategy and close the Kafka producer if needed, freeing resources
+     */
+    void stop();
 }

src/main/java/com/github/danielwegener/logback/kafka/delivery/DeliveryStrategy.java

@@ -0,0 +1,36 @@
+package com.github.danielwegener.logback.kafka.delivery;
+
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+import java.util.Map;
+
+/**
+ * {@link DeliveryStrategy} which initializes the Kafka producer on startup
+ * and sends messages asynchronously
+ */
+public class EarlyAsyncDeliveryStrategy<E,K,V> extends BaseDeliveryStrategy<E,K,V> {
+    private Producer<K, V> producer;
+
+    @Override
+    public void start(Map<String, Object> producerConfig, FailedDeliveryCallback<E> failedDeliveryCallback) {
+        super.start(producerConfig, failedDeliveryCallback);
+        producer = createProducer();
+    }
+
+    @Override
+    public Producer<K, V> getProducer() {
+        return producer;
+    }
+
+    @Override
+    public void send(ProducerRecord<K, V> record, E event) {
+        doSend(record, event);
+    }
+
+    @Override
+    public void stop() {
+        stopProducer();
+        producer = null;
+    }
+}