(feat) rename messageHandler to consumer record handler

Author: ylgrgyq

src/main/java/cn/leancloud/kafka/consumer/CatchAllExceptionConsumerRecordHandler.java

@@ -0,0 +1,60 @@
+package cn.leancloud.kafka.consumer;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.function.BiConsumer;
+
+/**
+ * A wrapper over a {@link ConsumerRecordHandler} to catch and swallow all the exceptions throw from the
+ * wrapped {@code ConsumerRecordHandler} when it failed to handle a consumed record.
+ * <p>
+ * This handler seems good to improve the availability of the consumer because it can swallow all the exceptions
+ * on handling a record and carry on to handle next record. But it actually can compromise
+ * the consumer to prevent a livelock, where the application did not crash but fails to
+ * make progress for some reason.
+ * Please use it judiciously. Usually fail fast, let the polling thread exit on exception, is your best choice.
+ *
+ * @param <K> the type of key for records consumed from Kafka
+ * @param <V> the type of value for records consumed from Kafka
+ */
+public final class CatchAllExceptionConsumerRecordHandler<K, V> implements ConsumerRecordHandler<K, V> {
+    private static final Logger logger = LoggerFactory.getLogger(CatchAllExceptionConsumerRecordHandler.class);
+
+    private final ConsumerRecordHandler<K, V> wrappedHandler;
+    private final BiConsumer<ConsumerRecord<K, V>, Throwable> errorConsumer;
+
+    /**
+     * Constructor for {@code CatchAllExceptionConsumerRecordHandler} to just log the failed record when
+     * the wrapped handler failed on calling {@link ConsumerRecordHandler#handleRecord(ConsumerRecord)}.
+     *
+     * @param wrappedHandler the wrapped {@link ConsumerRecordHandler}.
+     */
+    public CatchAllExceptionConsumerRecordHandler(ConsumerRecordHandler<K, V> wrappedHandler) {
+        this(wrappedHandler, (record, throwable) -> logger.error("Handle kafka consumer record: " + record + " failed.", throwable));
+    }
+
+    /**
+     * Constructor for {@code CatchAllExceptionConsumerRecordHandler} to use a {@link BiConsumer} to handle the
+     * failed record when the wrapped handler failed on calling {@link ConsumerRecordHandler#handleRecord(ConsumerRecord)}.
+     *
+     * @param wrappedHandler the wrapped {@link ConsumerRecordHandler}.
+     * @param errorConsumer  a {@link BiConsumer} to consume the failed record and the exception thrown from
+     *                       the {@link ConsumerRecordHandler#handleRecord(ConsumerRecord)}
+     */
+    public CatchAllExceptionConsumerRecordHandler(ConsumerRecordHandler<K, V> wrappedHandler,
+                                                  BiConsumer<ConsumerRecord<K, V>, Throwable> errorConsumer) {
+        this.wrappedHandler = wrappedHandler;
+        this.errorConsumer = errorConsumer;
+    }
+
+    @Override
+    public void handleRecord(ConsumerRecord<K, V> record) {
+        try {
+            wrappedHandler.handleRecord(record);
+        } catch (Exception ex) {
+            errorConsumer.accept(record, ex);
+        }
+    }
+}

src/main/java/cn/leancloud/kafka/consumer/ConsumerRecordHandler.java

@@ -0,0 +1,27 @@
+package cn.leancloud.kafka.consumer;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+/**
+ * A handler to handle all the consumer records consumed from Kafka broker.
+ *
+ * @param <K> the type of key for records consumed from Kafka
+ * @param <V> the type of value for records consumed from Kafka
+ */
+public interface ConsumerRecordHandler<K, V> {
+    /**
+     * Handle a {@link ConsumerRecord} consumed from Kafka broker.
+     * <p>
+     * For the sake of fail fast, if any exception throws from this method, the thread for fetching records
+     * from Kafka broker will exit with an error message. This will trigger rebalances of partitions
+     * among all the consumer groups this consumer joined and let other consumer try to handle the failed and the
+     * following records.
+     * <p>
+     * You can consider to use {@link RetriableConsumerRecordHandler} to retry handling process automatically.
+     * Or use {@link CatchAllExceptionConsumerRecordHandler} as a safety net to handle all the failed records in
+     * the same way.
+     *
+     * @param record the consumed {@link ConsumerRecord}
+     */
+    void handleRecord(ConsumerRecord<K, V> record);
+}

src/main/java/cn/leancloud/kafka/consumer/Fetcher.java

@@ -19,21 +19,21 @@ final class Fetcher<K, V> implements Runnable, Closeable {
 
     private final long pollTimeout;
     private final Consumer<K, V> consumer;
-    private final MessageHandler<K, V> handler;
+    private final ConsumerRecordHandler<K, V> handler;
     private final ExecutorCompletionService<ConsumerRecord<K, V>> service;
     private final Map<ConsumerRecord<K, V>, Future<ConsumerRecord<K, V>>> pendingFutures;
     private final CommitPolicy<K, V> policy;
     private final long gracefulShutdownMillis;
     private volatile boolean closed;
 
     Fetcher(LcKafkaConsumerBuilder<K, V> consumerBuilder) {
-        this(consumerBuilder.getConsumer(), consumerBuilder.getPollTimeout(), consumerBuilder.getMessageHandler(),
+        this(consumerBuilder.getConsumer(), consumerBuilder.getPollTimeout(), consumerBuilder.getConsumerRecordHandler(),
                 consumerBuilder.getWorkerPool(), consumerBuilder.getPolicy(), consumerBuilder.gracefulShutdownMillis());
     }
 
     Fetcher(Consumer<K, V> consumer,
             long pollTimeout,
-            MessageHandler<K, V> handler,
+            ConsumerRecordHandler<K, V> handler,
             ExecutorService workerPool,
             CommitPolicy<K, V> policy,
             long gracefulShutdownMillis) {
@@ -96,10 +96,10 @@ private boolean closed() {
     }
 
     private void dispatchFetchedRecords(ConsumerRecords<K, V> records) {
-        final MessageHandler<K, V> handler = this.handler;
+        final ConsumerRecordHandler<K, V> handler = this.handler;
         for (ConsumerRecord<K, V> record : records) {
             final Future<ConsumerRecord<K, V>> future = service.submit(() -> {
-                handler.handleMessage(record);
+                handler.handleRecord(record);
                 return record;
             });
             pendingFutures.put(record, future);

src/main/java/cn/leancloud/kafka/consumer/LcKafkaConsumer.java

@@ -7,6 +7,19 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 
+/**
+ * {@code LcKafkaConsumer} is a wrapper over {@link Consumer}. It will use {@link Consumer} to consume
+ * records from Kafka broker.
+ * <p>
+ * With {@link LcKafkaConsumer}, you can subscribe to several topics and handle all the records from these topic
+ * in a dedicated thread pool without warring polling timeout or session timeout due to the polling thread failed
+ * to poll spend too much time on process records
+ * <p>
+ * All the public methods in {@code LcKafkaConsumer} is thread safe.
+ *
+ * @param <K> the type of key for records consumed from Kafka
+ * @param <V> the type of value for records consumed from Kafka
+ */
 public final class LcKafkaConsumer<K, V> implements Closeable {
     enum State {
         INIT(0),
@@ -42,6 +55,13 @@ int code() {
         this.fetcherThread = new Thread(fetcher);
     }
 
+    /**
+     * Subscribe some Kafka topics to consume records from them.
+     *
+     * @param topics the topics to consume.
+     * @throws IllegalStateException    if this {@code LcKafkaConsumer} has closed or subscribed to some topics
+     * @throws IllegalArgumentException if the input {@code topics} is empty
+     */
     public synchronized void subscribe(Collection<String> topics) {
         if (topics.isEmpty()) {
             throw new IllegalArgumentException("subscribe empty topics");
@@ -95,7 +115,7 @@ boolean closed() {
         return state == State.CLOSED;
     }
 
-    CommitPolicy<K,V> policy() {
+    CommitPolicy<K, V> policy() {
         return policy;
     }
 }

src/main/java/cn/leancloud/kafka/consumer/LcKafkaConsumerBuilder.java

@@ -14,43 +14,50 @@
 import static cn.leancloud.kafka.consumer.BasicConsumerConfigs.ENABLE_AUTO_COMMIT;
 import static java.util.Objects.requireNonNull;
 
+/**
+ * A builder used to create a {@link LcKafkaConsumer} which uses a {@link KafkaConsumer} to consume records
+ * from Kafka broker.
+ *
+ * @param <K> the type of key for records consumed from Kafka
+ * @param <V> the type of value for records consumed from Kafka
+ */
 public final class LcKafkaConsumerBuilder<K, V> {
     /**
      * Create a {@code LcKafkaConsumerBuilder} used to build {@link LcKafkaConsumer}.
      *
-     * @param kafkaConfigs   the kafka consumer configs. Please refer
+     * @param kafkaConfigs   the kafka configs for {@link KafkaConsumer}. Please refer
      *                       <a href="http://kafka.apache.org/documentation.html#consumerconfigs" >this document</a> for
      *                       valid configurations.
-     * @param messageHandler a {@link MessageHandler} to handle the consumed msg from kafka
+     * @param consumerRecordHandler a {@link ConsumerRecordHandler} to handle the consumed record from kafka
      * @return a new {@code LcKafkaConsumerBuilder}
      */
     public static <K, V> LcKafkaConsumerBuilder<K, V> newBuilder(Map<String, Object> kafkaConfigs,
-                                                                 MessageHandler<K, V> messageHandler) {
+                                                                 ConsumerRecordHandler<K, V> consumerRecordHandler) {
         requireNonNull(kafkaConfigs, "kafkaConfigs");
-        requireNonNull(messageHandler, "messageHandler");
-        return new LcKafkaConsumerBuilder<>(new HashMap<>(kafkaConfigs), messageHandler);
+        requireNonNull(consumerRecordHandler, "consumerRecordHandler");
+        return new LcKafkaConsumerBuilder<>(new HashMap<>(kafkaConfigs), consumerRecordHandler);
     }
 
     /**
      * Create a {@code LcKafkaConsumerBuilder} used to build {@link LcKafkaConsumer}.
      *
-     * @param kafkaConfigs      the kafka consumer configs. Please refer
+     * @param kafkaConfigs      the kafka configs for {@link KafkaConsumer}. Please refer
      *                          <a href="http://kafka.apache.org/documentation.html#consumerconfigs" >this document</a> for
      *                          valid configurations.
-     * @param messageHandler    a {@link MessageHandler} to handle the consumed msg from kafka
+     * @param consumerRecordHandler    a {@link ConsumerRecordHandler} to handle the consumed record from kafka
      * @param keyDeserializer   The deserializer for key that implements {@link Deserializer}
      * @param valueDeserializer The deserializer for value that implements {@link Deserializer}
      * @return a new {@code LcKafkaConsumerBuilder}
      */
     public static <K, V> LcKafkaConsumerBuilder<K, V> newBuilder(Map<String, Object> kafkaConfigs,
-                                                                 MessageHandler<K, V> messageHandler,
+                                                                 ConsumerRecordHandler<K, V> consumerRecordHandler,
                                                                  Deserializer<K> keyDeserializer,
                                                                  Deserializer<V> valueDeserializer) {
         requireNonNull(kafkaConfigs, "kafkaConfigs");
-        requireNonNull(messageHandler, "messageHandler");
+        requireNonNull(consumerRecordHandler, "consumerRecordHandler");
         requireNonNull(keyDeserializer, "keyDeserializer");
         requireNonNull(valueDeserializer, "valueDeserializer");
-        return new LcKafkaConsumerBuilder<>(new HashMap<>(kafkaConfigs), messageHandler, keyDeserializer, valueDeserializer);
+        return new LcKafkaConsumerBuilder<>(new HashMap<>(kafkaConfigs), consumerRecordHandler, keyDeserializer, valueDeserializer);
     }
 
     /**
@@ -68,7 +75,7 @@ private static void requireArgument(boolean expression, String template, Object.
     private ExecutorService workerPool = ImmediateExecutorService.INSTANCE;
     private boolean shutdownWorkerPoolOnStop = false;
     private Map<String, Object> configs;
-    private MessageHandler<K, V> messageHandler;
+    private ConsumerRecordHandler<K, V> consumerRecordHandler;
     @Nullable
     private Consumer<K, V> consumer;
     @Nullable
@@ -79,18 +86,18 @@ private static void requireArgument(boolean expression, String template, Object.
     private CommitPolicy<K, V> policy;
 
     private LcKafkaConsumerBuilder(Map<String, Object> kafkaConsumerConfigs,
-                                   MessageHandler<K, V> messageHandler) {
-        this(kafkaConsumerConfigs, messageHandler, null, null);
+                                   ConsumerRecordHandler<K, V> consumerRecordHandler) {
+        this(kafkaConsumerConfigs, consumerRecordHandler, null, null);
     }
 
     private LcKafkaConsumerBuilder(Map<String, Object> kafkaConsumerConfigs,
-                                   MessageHandler<K, V> messageHandler,
+                                   ConsumerRecordHandler<K, V> consumerRecordHandler,
                                    @Nullable
-                                   Deserializer<K> keyDeserializer,
+                                           Deserializer<K> keyDeserializer,
                                    @Nullable
-                                   Deserializer<V> valueDeserializer) {
+                                           Deserializer<V> valueDeserializer) {
         this.configs = kafkaConsumerConfigs;
-        this.messageHandler = messageHandler;
+        this.consumerRecordHandler = consumerRecordHandler;
         this.keyDeserializer = keyDeserializer;
         this.valueDeserializer = valueDeserializer;
     }
@@ -131,13 +138,13 @@ public LcKafkaConsumerBuilder<K, V> pollTimeout(Duration pollTimeout) {
         return this;
     }
 
-    public LcKafkaConsumerBuilder<K,V> gracefulShutdownTimeoutMillis(long gracefulShutdownMs) {
+    public LcKafkaConsumerBuilder<K, V> gracefulShutdownTimeoutMillis(long gracefulShutdownMs) {
         requireArgument(gracefulShutdownMs >= 0, "gracefulShutdownMillis: %s (expected >= 0)", gracefulShutdownMs);
         this.gracefulShutdownMillis = gracefulShutdownMs;
         return this;
     }
 
-    public LcKafkaConsumerBuilder<K,V> gracefulShutdownTimeout(Duration gracefulShutdownTimeout) {
+    public LcKafkaConsumerBuilder<K, V> gracefulShutdownTimeout(Duration gracefulShutdownTimeout) {
         requireNonNull(gracefulShutdownTimeout, "gracefulShutdownTimeout");
         this.gracefulShutdownMillis = gracefulShutdownTimeout.toMillis();
         return this;
@@ -158,14 +165,14 @@ public LcKafkaConsumerBuilder<K, V> maxPendingAsyncCommits(int maxPendingAsyncCo
     }
 
     /**
-     * Change the {@link MessageHandler} to handle the consumed msg from kafka.
+     * Change the {@link ConsumerRecordHandler} to handle the consumed record from kafka.
      *
-     * @param messageHandler the handler to handle consumed msg
+     * @param consumerRecordHandler the handler to handle consumed record
      * @return this
      */
-    public LcKafkaConsumerBuilder<K, V> messageHandler(MessageHandler<K, V> messageHandler) {
-        requireNonNull(messageHandler, "messageHandler");
-        this.messageHandler = messageHandler;
+    public LcKafkaConsumerBuilder<K, V> messageHandler(ConsumerRecordHandler<K, V> consumerRecordHandler) {
+        requireNonNull(consumerRecordHandler, "consumerRecordHandler");
+        this.consumerRecordHandler = consumerRecordHandler;
         return this;
     }
 
@@ -207,7 +214,7 @@ public LcKafkaConsumerBuilder<K, V> workerPool(ExecutorService workerPool, boole
      * while (true) {
      *     final ConsumerRecords<K, V> records = consumer.poll(pollTimeout);
      *     for (ConsumerRecord<K, V> record : records) {
-     *         handler.handleMessage(record.topic(), record.value());
+     *         handler.handleRecord(record.topic(), record.value());
      *     }
      * }
      * </pre>
@@ -216,9 +223,9 @@ public LcKafkaConsumerBuilder<K, V> workerPool(ExecutorService workerPool, boole
      * Please note that this consumer requires these kafka configs must be set, otherwise
      * {@link IllegalArgumentException} will be thrown:
      * <ol>
-     *  <li><code>max.poll.interval.ms</code></li>
-     *  <li><code>max.poll.records</code></li>
-     *  <li><code>auto.commit.interval.ms</code></li>
+     * <li><code>max.poll.interval.ms</code></li>
+     * <li><code>max.poll.records</code></li>
+     * <li><code>auto.commit.interval.ms</code></li>
      * </ol>
      * <p>
      * Though all of these configs have default values in kafka, we still require every user to set them specifically.
@@ -266,8 +273,8 @@ Consumer<K, V> getConsumer() {
         return consumer;
     }
 
-    MessageHandler<K, V> getMessageHandler() {
-        return messageHandler;
+    ConsumerRecordHandler<K, V> getConsumerRecordHandler() {
+        return consumerRecordHandler;
     }
 
     ExecutorService getWorkerPool() {