(feat) graceful shutdown

Author: ylgrgyq

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

@@ -113,8 +113,8 @@ private LcKafkaConsumerBuilder(Map<String, Object> kafkaConsumerConfigs,
      * @param pollTimeoutMs the poll timeout in milliseconds
      * @return this
      */
-    public LcKafkaConsumerBuilder<K, V> pollTimeoutMs(long pollTimeoutMs) {
-        requireArgument(pollTimeoutMs >= 0, "pollTimeoutMs: %s (expect >= 0)", pollTimeoutMs);
+    public LcKafkaConsumerBuilder<K, V> pollTimeoutMillis(long pollTimeoutMs) {
+        requireArgument(pollTimeoutMs >= 0, "pollTimeoutMillis: %s (expect >= 0)", pollTimeoutMs);
         this.pollTimeout = pollTimeoutMs;
         return this;
     }
@@ -137,13 +137,13 @@ public LcKafkaConsumerBuilder<K, V> pollTimeout(Duration pollTimeout) {
         return this;
     }
 
-    public LcKafkaConsumerBuilder<K,V> gracefulShutdownMs(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> gracefulShutdownMs(Duration duration) {
+    public LcKafkaConsumerBuilder<K,V> gracefulShutdownTimeout(Duration duration) {
         requireNonNull(duration);
         this.gracefulShutdownMillis = duration.toMillis();
         return this;

src/test/java/cn/leancloud/kafka/consumer/LcKafkaConsumerBuilderTest.java

@@ -82,9 +82,9 @@ public void testNullDeserializers() {
     @Test
     public void testNegativePollTimeoutMs() {
         assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, testingHandler, keyDeserializer, valueDeserializer)
-                .pollTimeoutMs(-1 * ThreadLocalRandom.current().nextLong(1, Long.MAX_VALUE)))
+                .pollTimeoutMillis(-1 * ThreadLocalRandom.current().nextLong(1, Long.MAX_VALUE)))
                 .isInstanceOf(IllegalArgumentException.class)
-                .hasMessageContaining("pollTimeoutMs");
+                .hasMessageContaining("pollTimeoutMillis");
     }
 
     @Test
@@ -157,7 +157,7 @@ public void testAutoConsumerWithShouldShutdownWorkerPool() {
         configs.put("auto.offset.reset", "latest");
         assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, testingHandler, keyDeserializer, valueDeserializer)
                 .mockKafkaConsumer(new MockConsumer<>(OffsetResetStrategy.LATEST))
-                .pollTimeoutMs(1000)
+                .pollTimeoutMillis(1000)
                 .maxPendingAsyncCommits(100)
                 .workerPool(workerPool, true)
                 .buildAuto())
@@ -173,7 +173,7 @@ public void testAutoConsumer() {
         configs.put("auto.offset.reset", "latest");
         final LcKafkaConsumer<Object, Object> consumer = LcKafkaConsumerBuilder.newBuilder(configs, testingHandler)
                 .mockKafkaConsumer(new MockConsumer<>(OffsetResetStrategy.LATEST))
-                .pollTimeoutMs(1000)
+                .pollTimeoutMillis(1000)
                 .maxPendingAsyncCommits(100)
                 .workerPool(workerPool, false)
                 .buildAuto();

src/test/java/cn/leancloud/kafka/consumer/integration/Bootstrap.java

@@ -53,7 +53,7 @@ public LcKafkaConsumer<Integer, String> buildConsumer(String consumerName, TestS
             configs.put("auto.offset.reset", "earliest");
             configs.put("group.id", "2614911922612339122");
             configs.put("max.poll.records", 2);
-            configs.put("max.poll.interval.ms", "5000");
+            configs.put("max.poll.interval.ms", 5000);
 
             return LcKafkaConsumerBuilder.newBuilder(
                     configs,