(fix) partial async commit

Author: ylgrgyq

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

@@ -39,7 +39,6 @@ public Set<TopicPartition> tryCommit(boolean noPendingRecords) {
                 if (exception != null) {
                     logger.warn("Failed to commit offsets: " + offsets + " asynchronously", exception);
                     forceSync = true;
-
                 }
             });
         }

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

@@ -143,9 +143,9 @@ public LcKafkaConsumerBuilder<K,V> gracefulShutdownTimeoutMillis(long gracefulSh
         return this;
     }
 
-    public LcKafkaConsumerBuilder<K,V> gracefulShutdownTimeout(Duration duration) {
-        requireNonNull(duration);
-        this.gracefulShutdownMillis = duration.toMillis();
+    public LcKafkaConsumerBuilder<K,V> gracefulShutdownTimeout(Duration gracefulShutdownTimeout) {
+        requireNonNull(gracefulShutdownTimeout, "gracefulShutdownTimeout");
+        this.gracefulShutdownMillis = gracefulShutdownTimeout.toMillis();
         return this;
     }
 

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

@@ -44,16 +44,18 @@ public Set<TopicPartition> tryCommit(boolean noPendingRecords) {
             consumer.commitAsync(completedTopicOffsets, (offsets, exception) -> {
                 --pendingAsyncCommitCounter;
                 assert pendingAsyncCommitCounter >= 0 : "actual: " + pendingAsyncCommitCounter;
-                final Map<TopicPartition, OffsetAndMetadata> completeOffsets =
-                        offsets == completedTopicOffsets ? new HashMap<>(offsets) : offsets;
-                for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : completeOffsets.entrySet()) {
-                    completedTopicOffsets.remove(entry.getKey(), entry.getValue());
-                    topicOffsetHighWaterMark.remove(entry.getKey(), entry.getValue().offset());
-                }
-
                 if (exception != null) {
+                    // if last async commit is failed, we do not clean cached completed offsets and let next
+                    // commit be a sync commit so all the complete offsets will be committed at that time
                     logger.warn("Failed to commit offset: " + offsets + " asynchronously", exception);
                     forceSync = true;
+                } else {
+                    final Map<TopicPartition, OffsetAndMetadata> completeOffsets =
+                            offsets == completedTopicOffsets ? new HashMap<>(offsets) : offsets;
+                    for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : completeOffsets.entrySet()) {
+                        completedTopicOffsets.remove(entry.getKey(), entry.getValue());
+                        topicOffsetHighWaterMark.remove(entry.getKey(), entry.getValue().offset());
+                    }
                 }
             });
         }

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

@@ -87,6 +87,30 @@ public void testNegativePollTimeoutMs() {
                 .hasMessageContaining("pollTimeoutMillis");
     }
 
+    @Test
+    public void testNullPollTimeout() {
+        assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, testingHandler, keyDeserializer, valueDeserializer)
+                .pollTimeout(null))
+                .isInstanceOf(NullPointerException.class)
+                .hasMessage("pollTimeout");
+    }
+
+    @Test
+    public void testNegativeShutdownTimeout() {
+        assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, testingHandler, keyDeserializer, valueDeserializer)
+                .gracefulShutdownTimeoutMillis(-1 * ThreadLocalRandom.current().nextLong(1, Long.MAX_VALUE)))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("gracefulShutdownMillis");
+    }
+
+    @Test
+    public void testNullShutdownTimeout() {
+        assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, testingHandler, keyDeserializer, valueDeserializer)
+                .gracefulShutdownTimeout(null))
+                .isInstanceOf(NullPointerException.class)
+                .hasMessage("gracefulShutdownTimeout");
+    }
+
     @Test
     public void testNegativeMaxPendingAsyncCommits() {
         assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, testingHandler, keyDeserializer, valueDeserializer)
@@ -103,14 +127,6 @@ public void testZeroMaxPendingAsyncCommits() {
                 .hasMessageContaining("maxPendingAsyncCommits");
     }
 
-    @Test
-    public void testNullPollTimeout() {
-        assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, testingHandler, keyDeserializer, valueDeserializer)
-                .pollTimeout(null))
-                .isInstanceOf(NullPointerException.class)
-                .hasMessage("pollTimeout");
-    }
-
     @Test
     public void testNullWorkerPool() {
         assertThatThrownBy(() -> LcKafkaConsumerBuilder.newBuilder(configs, testingHandler, keyDeserializer, valueDeserializer)

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

@@ -168,6 +168,7 @@ public void testForceSyncAfterAsyncCommitFailed() {
 
         policy = new PartialAsyncCommitPolicy<>(mockConsumer, 10);
 
+        // a failed async commit on the first time
         final ConsumerRecord<Object, Object> triggerFailedRecord = pendingRecords.get(0);
         completeRecord(triggerFailedRecord);
         assertThat(policy.tryCommit(true))
@@ -177,11 +178,15 @@ public void testForceSyncAfterAsyncCommitFailed() {
         verify(mockConsumer, times(1)).commitAsync(any(), any());
         verify(mockConsumer, never()).commitSync(any());
 
+        // sync commit after the failed async commit
         final ConsumerRecord<Object, Object> syncRecord = pendingRecords.get(1);
         completeRecord(syncRecord);
         assertThat(policy.tryCommit(true))
-                .hasSize(1)
-                .isEqualTo(Collections.singleton(new TopicPartition(syncRecord.topic(), syncRecord.partition())));
+                .hasSize(2)
+                .containsExactlyInAnyOrderElementsOf(Arrays.asList(
+                        new TopicPartition(syncRecord.topic(), syncRecord.partition()),
+                        new TopicPartition(syncRecord.topic(), triggerFailedRecord.partition())
+                ));
 
         verify(mockConsumer, times(1)).commitAsync(any(), any());
         verify(mockConsumer, times(1)).commitSync(any());