[GitHub] merlimat closed pull request #852: Avoid interrupted exception in PulsarKafkaConsumer when shutting down

[GitBox <gi...@apache.org>]

merlimat closed pull request #852: Avoid interrupted exception in PulsarKafkaConsumer when shutting down
URL: https://github.com/apache/incubator-pulsar/pull/852
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java
index fcd40b6ef..6ece7fa0f 100644
--- a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java
@@ -60,8 +60,6 @@
 import org.apache.pulsar.client.util.ConsumerName;
 import org.apache.pulsar.client.util.FutureUtil;
 import org.apache.pulsar.common.naming.DestinationName;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import com.google.common.collect.Lists;
 
@@ -82,6 +80,8 @@
     private final Map<TopicPartition, Long> lastReceivedOffset = new ConcurrentHashMap<>();
     private final Map<TopicPartition, OffsetAndMetadata> lastCommittedOffset = new ConcurrentHashMap<>();
 
+    private volatile boolean closed = false;
+
     private static class QueueItem {
         final org.apache.pulsar.client.api.Consumer consumer;
         final Message message;
@@ -159,7 +159,12 @@ public void received(org.apache.pulsar.client.api.Consumer consumer, Message msg
         try {
             receivedMessages.put(new QueueItem(consumer, msg));
         } catch (InterruptedException e) {
-            throw new RuntimeException(e);
+            Thread.currentThread().interrupt();
+            if (closed) {
+                // Consumer was closed and the thread was interrupted. Nothing to worry about here
+            } else {
+                throw new RuntimeException(e);
+            }
         }
     }
 
@@ -512,6 +517,8 @@ public void close() {
     @Override
     public void close(long timeout, TimeUnit unit) {
         try {
+            closed = true;
+
             if (isAutoCommit) {
                 commitAsync();
             }
@@ -526,5 +533,4 @@ public void wakeup() {
         throw new UnsupportedOperationException();
     }
 
-    private static final Logger log = LoggerFactory.getLogger(PulsarKafkaConsumer.class);
 }
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index fb9efdd91..1c950b3a2 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -309,8 +309,14 @@ protected Message internalReceive(int timeout, TimeUnit unit) throws PulsarClien
             return message;
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
-            stats.incrementNumReceiveFailed();
-            throw new PulsarClientException(e);
+
+            State state = getState();
+            if (state != State.Closing && state != State.Closed) {
+                stats.incrementNumReceiveFailed();
+                throw new PulsarClientException(e);
+            } else {
+                return null;
+            }
         }
     }
 


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services