[GitHub] [pulsar] eolivelli commented on a change in pull request #10352: Make Consumer thread safe and lock-free

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

eolivelli commented on a change in pull request #10352:
URL: https://github.com/apache/pulsar/pull/10352#discussion_r620871200



##########
File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
##########
@@ -957,10 +951,12 @@ private void closeConsumerTasks() {
     }
 
     private void failPendingReceive() {
-        if (pinnedExecutor != null && !pinnedExecutor.isShutdown()) {
-            failPendingReceives(this.pendingReceives);
-            failPendingBatchReceives(this.pendingBatchReceives);
-        }
+        internalPinnedExecutor.execute(() -> {
+            if (pinnedExecutor != null && !pinnedExecutor.isShutdown()) {

Review comment:
       Testing for IaShutdown is probably not enough, because in the meantime someone may shutdown it.
   Probably we can add try/catch RejectedExecutionException 

##########
File path: pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java
##########
@@ -48,6 +47,7 @@
         when(clientMock.timer()).thenReturn(mock(Timer.class));
 
         when(clientMock.externalExecutorProvider()).thenReturn(mock(ExecutorProvider.class));
+        when(clientMock.getInternalExecutorService()).thenReturn(Executors.newSingleThreadExecutor());
         when(clientMock.eventLoopGroup().next()).thenReturn(mock(EventLoop.class));

Review comment:
       Please shutdown this executor




-- 
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.

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