(chore) more docs

ylgrgyq · ylgrgyq · commit 3ec88f3ad333 · 2019-12-27T17:58:15.000+08:00
diff --git a/src/main/java/cn/leancloud/kafka/consumer/ImmediateExecutorService.java b/src/main/java/cn/leancloud/kafka/consumer/ImmediateExecutorService.java
@@ -6,7 +6,9 @@
 import java.util.concurrent.TimeUnit;
 
 /**
- * This is an shared {@link java.util.concurrent.ExecutorService} so it can not be shutdown
+ * A {@link java.util.concurrent.ExecutorService} which run all the tasks in the thread which submit the task.
+ * <p>
+ * This is an shared {@link java.util.concurrent.ExecutorService} so it can not be shutdown.
  */
 final class ImmediateExecutorService extends AbstractExecutorService {
     static final ImmediateExecutorService INSTANCE = new ImmediateExecutorService();
diff --git a/src/main/java/cn/leancloud/kafka/consumer/LcKafkaConsumer.java b/src/main/java/cn/leancloud/kafka/consumer/LcKafkaConsumer.java
@@ -11,9 +11,9 @@
  * {@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
+ * With {@link LcKafkaConsumer}, you can subscribe to several topics and handle all the records from these topics
+ * in a dedicated thread pool without warring polling timeout or session timeout due to the polling thread spend
+ * too much time on process records and failed to poll broker at least once within {@code max.poll.interval.ms}.
  * <p>
  * All the public methods in {@code LcKafkaConsumer} is thread safe.
  *
diff --git a/src/main/java/cn/leancloud/kafka/consumer/LcKafkaConsumerBuilder.java b/src/main/java/cn/leancloud/kafka/consumer/LcKafkaConsumerBuilder.java
@@ -25,9 +25,9 @@ public final class LcKafkaConsumerBuilder<K, V> {
     /**
      * Create a {@code LcKafkaConsumerBuilder} used to build {@link LcKafkaConsumer}.
      *
-     * @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 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 consumerRecordHandler a {@link ConsumerRecordHandler} to handle the consumed record from kafka
      * @return a new {@code LcKafkaConsumerBuilder}
      */
@@ -41,12 +41,12 @@ public static <K, V> LcKafkaConsumerBuilder<K, V> newBuilder(Map<String, Object>
     /**
      * Create a {@code LcKafkaConsumerBuilder} used to build {@link LcKafkaConsumer}.
      *
-     * @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 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}
+     * @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 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,
@@ -127,7 +127,6 @@ public LcKafkaConsumerBuilder<K, V> pollTimeoutMillis(long pollTimeoutMs) {
      * If 0, poll operation will return immediately with any records that are available currently in the buffer,
      * else returns empty.
      * <p>
-     * Must not be negative.
      *
      * @param pollTimeout the poll timeout duration
      * @return this
@@ -138,12 +137,27 @@ public LcKafkaConsumerBuilder<K, V> pollTimeout(Duration pollTimeout) {
         return this;
     }
 
-    public LcKafkaConsumerBuilder<K, V> gracefulShutdownTimeoutMillis(long gracefulShutdownMs) {
-        requireArgument(gracefulShutdownMs >= 0, "gracefulShutdownMillis: %s (expected >= 0)", gracefulShutdownMs);
-        this.gracefulShutdownMillis = gracefulShutdownMs;
+    /**
+     * Sets the amount of time to wait after calling {@link LcKafkaConsumer#close()} for
+     * consumed records to handle before actually shutting down.
+     *
+     * @param gracefulShutdownTimeoutMillis the graceful shutdown timeout in milliseconds
+     * @return this
+     */
+    public LcKafkaConsumerBuilder<K, V> gracefulShutdownTimeoutMillis(long gracefulShutdownTimeoutMillis) {
+        requireArgument(gracefulShutdownTimeoutMillis >= 0,
+                "gracefulShutdownTimeoutMillis: %s (expected >= 0)", gracefulShutdownTimeoutMillis);
+        this.gracefulShutdownMillis = gracefulShutdownTimeoutMillis;
         return this;
     }
 
+    /**
+     * Sets the amount of time to wait after calling {@link LcKafkaConsumer#close()} for
+     * consumed records to handle before actually shutting down.
+     *
+     * @param gracefulShutdownTimeout the graceful shutdown timeout duration
+     * @return this
+     */
     public LcKafkaConsumerBuilder<K, V> gracefulShutdownTimeout(Duration gracefulShutdownTimeout) {
         requireNonNull(gracefulShutdownTimeout, "gracefulShutdownTimeout");
         this.gracefulShutdownMillis = gracefulShutdownTimeout.toMillis();
@@ -152,7 +166,8 @@ public LcKafkaConsumerBuilder<K, V> gracefulShutdownTimeout(Duration gracefulShu
 
     /**
      * When using async consumer to commit offset asynchronously, this argument can force consumer to do a synchronous
-     * commit after there's already {@code maxPendingAsyncCommits} async commits on the fly without response from broker.
+     * commit after there's already this ({@code maxPendingAsyncCommits}) many async commits on the fly without
+     * response from broker.
      *
      * @param maxPendingAsyncCommits do a synchronous commit when pending async commits beyond this limit
      * @return this
@@ -164,38 +179,45 @@ public LcKafkaConsumerBuilder<K, V> maxPendingAsyncCommits(int maxPendingAsyncCo
         return this;
     }
 
-    /**
-     * Change the {@link ConsumerRecordHandler} to handle the consumed record from kafka.
-     *
-     * @param consumerRecordHandler the handler to handle consumed record
-     * @return this
-     */
-    public LcKafkaConsumerBuilder<K, V> messageHandler(ConsumerRecordHandler<K, V> consumerRecordHandler) {
-        requireNonNull(consumerRecordHandler, "consumerRecordHandler");
-        this.consumerRecordHandler = consumerRecordHandler;
-        return this;
-    }
-
     /**
      * Internal testing usage only.
-     * Passing a {@link Consumer} as as the underlying kafka consumer. Usually this would be a {@link MockConsumer}.
+     * <p>
+     * Passing a {@link Consumer} as the underlying {@link Consumer}. Usually this would be a {@link MockConsumer}.
      *
+     * @param mockedConsumer the injected consumer
      * @return this
      */
-    LcKafkaConsumerBuilder<K, V> mockKafkaConsumer(Consumer<K, V> consumer) {
-        requireNonNull(consumer, "cn/leancloud/kafka/consumer");
-        if (consumer instanceof KafkaConsumer) {
+    LcKafkaConsumerBuilder<K, V> mockKafkaConsumer(Consumer<K, V> mockedConsumer) {
+        requireNonNull(mockedConsumer, "consumer");
+        if (mockedConsumer instanceof KafkaConsumer) {
             throw new IllegalArgumentException("need a mocked Consumer");
         }
-        this.consumer = consumer;
+        this.consumer = mockedConsumer;
         return this;
     }
 
     /**
-     * The thread pool used by consumer to handle the consumed messages from kafka. Please note that if you are
-     * using auto commit consumer, this thread pool is not be used.
+     * The thread pool used by consumer to handle the consumed records from Kafka broker. If no worker pool is provided,
+     * the created {@link LcKafkaConsumer} will use {@link ImmediateExecutorService} to handle records in
+     * the records polling thread instead.
+     * <p>
+     * When a worker pool is provided, after each poll, the polling thread will take one thread from this worker pool
+     * for each polled {@link org.apache.kafka.clients.consumer.ConsumerRecord} to handle the record. Please tune
+     * the <code>max.poll.records</code> in kafka configs to limit the number of records polled at each time do not
+     * exceed the max size of the provided worker thread pool. Otherwise, a
+     * {@link java.util.concurrent.RejectedExecutionException} will thrown when the polling thread submitting too much
+     * tasks to the pool. Then this exception will lead the only polling thread to exit.
+     * <p>
+     * If you are using partial sync/async commit consumer by building {@link LcKafkaConsumer} with
+     * {@link LcKafkaConsumerBuilder#buildPartialSync()} or {@link LcKafkaConsumerBuilder#buildPartialAsync()}, without
+     * a worker pool, they degrade to sync/async commit consumer as built with {@link LcKafkaConsumerBuilder#buildSync()}
+     * or {@link LcKafkaConsumerBuilder#buildAsync()}.
+     * <p>
+     * If no worker pool provided, you also need to tune {@code max.poll.interval.ms} in kafka configs, to ensure the
+     * polling thread can at least poll once within {@code max.poll.interval.ms} during handling consumed messages
+     * to prevent itself from session timeout or polling timeout.
      *
-     * @param workerPool     a thread pool to handle consumed messages
+     * @param workerPool     a thread pool to handle consumed records
      * @param shutdownOnStop true to shutdown the input worker pool when this consumer closed
      * @return this
      */
@@ -207,33 +229,23 @@ public LcKafkaConsumerBuilder<K, V> workerPool(ExecutorService workerPool, boole
     }
 
     /**
-     * Build a consumer which commit offset automatically in fixed interval. This consumer will not using other thread
-     * pool to handle consumed messages and will not pause any partition when after polling. This consumer is
-     * equivalent to:
-     * <pre>
-     * while (true) {
-     *     final ConsumerRecords<K, V> records = consumer.poll(pollTimeout);
-     *     for (ConsumerRecord<K, V> record : records) {
-     *         handler.handleRecord(record.topic(), record.value());
-     *     }
-     * }
-     * </pre>
-     *
+     * Build a consumer which commits offset automatically at fixed interval. It is both OK for with or without a
+     * worker thread pool. But without a worker pool, please tune the {@code max.poll.interval.ms} in
+     * Kafka configs as mentioned in {@link LcKafkaConsumerBuilder#workerPool(ExecutorService, boolean)}.
      * <p>
-     * Please note that this consumer requires these kafka configs must be set, otherwise
+     * This kind of consumer requires the following 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.offset.reset</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.
-     * Because these configs is vital for using this consumer safely. You should tune them to ensure the polling thread
-     * in this consumer can at least poll once within {@code max.poll.interval.ms} during handling consumed messages
-     * to prevent itself from session timeout.
+     * Because these configs is vital for using this consumer safely.
      * <p>
-     * Note that if you set {@code enable.auto.commit} to false, this consumer will set it to true by itself.
+     * If you set {@code enable.auto.commit} to false, this consumer will set it to true by itself.
      *
      * @return this
      */
@@ -244,24 +256,114 @@ public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildAuto() {
         return doBuild();
     }
 
+    /**
+     * Build a consumer in which the polling thread always does a sync commit after all the polled records has been handled.
+     * Because it only commits after all the polled records handled, so the longer the records handling process，
+     * the longer the interval between each commits, the bigger of the possibility to repeatedly consume a same record
+     * when the consumer crash.
+     * <p>
+     * This kind of consumer ensures to do a sync commit to commit all the finished records at that time when the
+     * consumer is shutdown or any partition was revoked. It requires the following kafka configs must be set,
+     * otherwise an {@link IllegalArgumentException} will be thrown:
+     * <ol>
+     *  <li><code>max.poll.records</code></li>
+     *  <li><code>auto.offset.reset</code></li>
+     * </ol>
+     * <p>
+     * Though all of these configs have default values in kafka, we still require every user to set them specifically.
+     * Because these configs is vital for using this consumer safely.
+     * <p>
+     * If you set {@code enable.auto.commit} to true, this consumer will set it to false by itself.
+     *
+     * @return this
+     */
     public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildSync() {
         consumer = buildConsumer(false);
         policy = new SyncCommitPolicy<>(consumer);
         return doBuild();
     }
 
+    /**
+     * Build a consumer in which the polling thread does a sync commits whenever there's any handled consumer records. It
+     * commits often, so after a consumer crash, comparatively little records may be handled more than once. But also
+     * due to commit often, the overhead causing by committing is relatively high.
+     * <p>
+     * This kind of consumer ensures to do a sync commit to commit all the finished records at that time when the
+     * consumer is shutdown or any partition was revoked. It requires the following kafka configs must be set,
+     * otherwise an {@link IllegalArgumentException} will be thrown:
+     * <ol>
+     *  <li><code>max.poll.records</code></li>
+     *  <li><code>auto.offset.reset</code></li>
+     * </ol>
+     * <p>
+     * Though all of these configs have default values in kafka, we still require every user to set them specifically.
+     * Because these configs is vital for using this consumer safely.
+     * <p>
+     * If you set {@code enable.auto.commit} to true, this consumer will set it to false by itself.
+     *
+     * @return this
+     */
     public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildPartialSync() {
         consumer = buildConsumer(false);
         policy = new PartialSyncCommitPolicy<>(consumer);
         return doBuild();
     }
 
+    /**
+     * Build a consumer in which the polling thread always does a async commit after all the polled records has been handled.
+     * Because it only commits after all the polled records handled, so the longer the records handling process，
+     * the longer the interval between each commits, the bigger of the possibility to repeatedly consume a same record
+     * when the consumer crash.
+     * <p>
+     * If any async commit is failed or the number of pending async commits is beyond the limit set by
+     * {@link LcKafkaConsumerBuilder#maxPendingAsyncCommits(int)}, this consumer will do a sync commit to commit all the
+     * records which have been handled.
+     * <p>
+     * This kind of consumer ensures to do a sync commit to commit all the finished records at that time when the
+     * consumer is shutdown or any partition was revoked. It requires the following kafka configs must be set,
+     * otherwise an {@link IllegalArgumentException} will be thrown:
+     * <ol>
+     *  <li><code>max.poll.records</code></li>
+     *  <li><code>auto.offset.reset</code></li>
+     * </ol>
+     * <p>
+     * Though all of these configs have default values in kafka, we still require every user to set them specifically.
+     * Because these configs is vital for using this consumer safely.
+     * <p>
+     * If you set {@code enable.auto.commit} to true, this consumer will set it to false by itself.
+     *
+     * @return this
+     */
     public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildAsync() {
         consumer = buildConsumer(false);
         policy = new AsyncCommitPolicy<>(consumer, maxPendingAsyncCommits);
         return doBuild();
     }
 
+    /**
+     * Build a consumer in which the polling thread does a async commits whenever there's any handled consumer records. It
+     * commits often, so after a consumer crash, comparatively little records may be handled more than once. It use
+     * async commit to mitigate the overhead causing by high committing times.
+     * <p>
+     * If any async commit is failed or the number of pending async commits is beyond the limit set by
+     * {@link LcKafkaConsumerBuilder#maxPendingAsyncCommits(int)}, this consumer will do a sync commit to commit all the
+     * records which have been handled.
+     * <p>
+     * This kind of consumer ensures to do a sync commit to commit all the finished records at that time when the
+     * consumer is shutdown or any partition was revoked. It requires the following kafka configs must be set,
+     * otherwise an {@link IllegalArgumentException} will be thrown:
+     * <ol>
+     *  <li><code>max.poll.records</code></li>
+     *  <li><code>auto.offset.reset</code></li>
+     * </ol>
+     * <p>
+     * Though all of these configs have default values in kafka, we still require every user to set them specifically.
+     * Because these configs is vital for using this consumer safely.
+     * <p>
+     * If you set {@code enable.auto.commit} to true, this consumer will set it to false by itself.
+     *
+     * @return this
+     */
     public <K1 extends K, V1 extends V> LcKafkaConsumer<K1, V1> buildPartialAsync() {
         consumer = buildConsumer(false);
         policy = new PartialAsyncCommitPolicy<>(consumer, maxPendingAsyncCommits);
diff --git a/src/test/java/cn/leancloud/kafka/consumer/LcKafkaConsumerBuilderTest.java b/src/test/java/cn/leancloud/kafka/consumer/LcKafkaConsumerBuilderTest.java
@@ -57,22 +57,6 @@ public void testNullKafkaConfigs() {
                 .hasMessage("kafkaConfigs");
     }
 
-    @Test
-    public void testNullMessageHandler() {
-        assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, null))
-                .isInstanceOf(NullPointerException.class)
-                .hasMessage("consumerRecordHandler");
-
-        assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, null, keyDeserializer, valueDeserializer))
-                .isInstanceOf(NullPointerException.class)
-                .hasMessage("consumerRecordHandler");
-
-        assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, testingHandler, keyDeserializer, valueDeserializer)
-                .messageHandler(null))
-                .isInstanceOf(NullPointerException.class)
-                .hasMessage("consumerRecordHandler");
-    }
-
     @Test
     public void testNullDeserializers() {
         assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, testingHandler, null, valueDeserializer))
@@ -104,7 +88,7 @@ public void testNegativeShutdownTimeout() {
         assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, testingHandler, keyDeserializer, valueDeserializer)
                 .gracefulShutdownTimeoutMillis(-1 * ThreadLocalRandom.current().nextLong(1, Long.MAX_VALUE)))
                 .isInstanceOf(IllegalArgumentException.class)
-                .hasMessageContaining("gracefulShutdownMillis");
+                .hasMessageContaining("gracefulShutdownTimeoutMillis");
     }
 
     @Test
