(fix) retrieable handler

Author: ylgrgyq

pom.xml

@@ -174,9 +174,6 @@
                 <version>3.0.1</version>
                 <configuration>
                     <source>8</source>
-                    <!-- sorry for get rid of doc warnings because we lack so much of docs... :tada -->
-                    <additionalOptions>-Xdoclint:none</additionalOptions>
-                    <additionalJOption>-Xdoclint:none</additionalJOption>
                 </configuration>
                 <executions>
                     <execution>

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

@@ -85,6 +85,8 @@ public synchronized void subscribe(Collection<String> topics) {
 
     /**
      * Get the metrics kept by the consumer
+     *
+     * @return the metrics kept by the consumer
      */
     public Map<MetricName, ? extends Metric> metrics() {
         return consumer.metrics();

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

@@ -53,21 +53,22 @@ public RetriableConsumerRecordHandler(ConsumerRecordHandler<K, V> wrappedHandler
     @Override
     public void handleRecord(ConsumerRecord<K, V> record) {
         Exception lastException = null;
-        int tried = 0;
-        while (tried++ < maxRetryTimes) {
+        int retried = 0;
+        while (retried <= maxRetryTimes) {
             try {
                 wrappedHandler.handleRecord(record);
                 return;
             } catch (Exception ex) {
                 lastException = ex;
-                if (tried != maxRetryTimes) {
-                    try {
-                        Thread.sleep(retryInterval.toMillis());
-                    } catch (InterruptedException e) {
-                        Thread.currentThread().interrupt();
-                        break;
-                    }
+                try {
+                    Thread.sleep(retryInterval.toMillis());
+                } catch (InterruptedException e) {
+                    // keep the interrupt status and still retry for the next time
+                    // because interrupt means we can't block this thread, but
+                    // it does not mean we should quit our job
+                    Thread.currentThread().interrupt();
                 }
+                ++retried;
             }
         }
 

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

@@ -34,7 +34,7 @@ public void testInvalidRetryTimes() {
     @Test
     public void testRetry() {
         final Exception expectedEx = new RuntimeException();
-        final int retryTimes = 10;
+        final int retryTimes = 9;
 
         doThrow(expectedEx).when(wrappedHandler).handleRecord(testingRecord);
 
@@ -44,7 +44,7 @@ public void testRetry() {
                 .isInstanceOf(HandleMessageFailedException.class)
                 .hasCause(expectedEx);
 
-        verify(wrappedHandler, times(retryTimes)).handleRecord(testingRecord);
+        verify(wrappedHandler, times(retryTimes + 1)).handleRecord(testingRecord);
     }
 
     @Test
@@ -61,8 +61,8 @@ public void testNoRetry() {
     }
 
     @Test
-    public void testRetrySuccess() {
-        final Exception expectedEx = new RuntimeException();
+    public void testRetryToSuccess() {
+        final Exception expectedEx = new RuntimeException("intended exception");
         final int retryTimes = 10;
 
         doThrow(expectedEx).doNothing().when(wrappedHandler).handleRecord(testingRecord);
@@ -75,7 +75,7 @@ public void testRetrySuccess() {
     }
 
     @Test
-    public void testInterruptInRetryInterval() throws Exception {
+    public void testInterruptInRetryIntervalAndFailForever() throws Exception {
         final Exception expectedEx = new RuntimeException();
         final int retryTimes = 10;
 
@@ -87,8 +87,23 @@ public void testInterruptInRetryInterval() throws Exception {
                 .isInstanceOf(HandleMessageFailedException.class)
                 .hasCause(expectedEx);
 
-        assertThat(Thread.currentThread().isInterrupted()).isTrue();
-        verify(wrappedHandler, times(1)).handleRecord(testingRecord);
+        assertThat(Thread.interrupted()).isTrue();
+        verify(wrappedHandler, times(retryTimes + 1)).handleRecord(testingRecord);
+    }
+
+    @Test
+    public void testInterruptInRetryIntervalAndRetryToSuccess() throws Exception {
+        final Exception expectedEx = new RuntimeException("intended exception");
+        final int retryTimes = 2;
+
+        doThrow(expectedEx).doThrow(expectedEx).doNothing().when(wrappedHandler).handleRecord(testingRecord);
+
+        final ConsumerRecordHandler<Object, Object> handler = new RetriableConsumerRecordHandler<>(wrappedHandler, retryTimes, Duration.ofHours(1));
+        interruptAfter(Thread.currentThread(), 200);
+        handler.handleRecord(testingRecord);
+
+        assertThat(Thread.interrupted()).isTrue();
+        verify(wrappedHandler, times(3)).handleRecord(testingRecord);
     }
 
     private void interruptAfter(Thread threadToInterrupt, long triggerDelayMillis) throws Exception {