Re: [I] [Java client] Consumer#receive returns null in case of InterruptedException and does not reset Thread.interrupted flag [pulsar]

["BewareMyPower (via GitHub)" <gi...@apache.org>]

BewareMyPower commented on issue #9921:
URL: https://github.com/apache/pulsar/issues/9921#issuecomment-3101801309

   I'm revisiting how existing synchronous methods handle the interruption and found this issue.
   
   > if you execute Consumer#receive() and thread is Interrupted the method returns a null value but it does not set Thread.currentThead().interrupt().
   
   @eolivelli It's confusing. Even if we didn't set the interruption status, the `receive` method (with timeout) still wouldn't return `null` for interruption.
   
   ```java
       @DataProvider
       public static Object[][] receiveWithTimeout() {
           return new Object[][] { { true }, { false } };
       }
   
       @Test(timeOut = 30000, dataProvider = "receiveWithTimeout")
       public void testReceiveInterrupted(boolean receiveWithTimeout) throws Exception {
           final String topic = "persistent://my-property/my-ns/testReceiveInterrupted";
           final var executor = Executors.newSingleThreadExecutor();
           try {
               final var consumerCreated = new CountDownLatch(1);
               final var taskFuture = new CompletableFuture<Void>();
               final var future = executor.submit(() -> {
                   try {
                       @Cleanup final var consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub")
                               .subscribe();
                       consumerCreated.countDown();
   
                       try {
                           if (receiveWithTimeout) {
                               consumer.receive(5, TimeUnit.SECONDS); // [1]
                           } else {
                               consumer.receive(); // it will be interrupted
                           }
                           taskFuture.completeExceptionally(new IllegalStateException("The 1st receive should fail"));
                       } catch (PulsarClientException e) {
                           if (!(e.getCause() instanceof InterruptedException)) {
                               log.error("The initial receive failed", e);
                               throw e;
                           }
                       }
   
                       for (int i = 0; i < 2; i++) {
                           try {
                               if (receiveWithTimeout) {
                                   consumer.receive(1, TimeUnit.SECONDS); // [2]
                               } else {
                                   consumer.receive();
                               }
                               taskFuture.completeExceptionally(
                                       new IllegalStateException("The subsequent receive " + i + " should fail"));
                           } catch (PulsarClientException e) {
                               if (!(e.getCause() instanceof InterruptedException)) {
                                   log.error("[{}] The subsequent receive failed", i, e);
                                   throw e;
                               }
                           }
                       }
                       taskFuture.complete(null);
                   } catch (Throwable e) {
                       taskFuture.completeExceptionally(e);
                   }
               });
               consumerCreated.await();
               future.cancel(true);
               taskFuture.get();
           } finally {
               executor.shutdown();
           }
   
       }
   ```
   
   I wrote the test above to verify the current behavior. However, even after I removed `ExceptionHandler.handleInterruptedException(e);` from `internalReceive`. The 1st `receive` call (`[1]`) still threw a `PulsarClientException` that wraps the `InterruptedException`. The difference is that the following `receive` call (`[2]`) wouldn't fail.
   
   /cc @lhotari 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org