pulsar-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [pulsar] branch master updated: Handle RejectedExecutionException at closing ManagedLedgerClientFactory (#2646)
Date Tue, 25 Sep 2018 02:01:30 GMT
This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 371f1db  Handle RejectedExecutionException at closing ManagedLedgerClientFactory
(#2646)
371f1db is described below

commit 371f1db8598be6b6ce11dfbadf7525e0e71e802f
Author: Sijie Guo <guosijie@gmail.com>
AuthorDate: Mon Sep 24 19:01:26 2018 -0700

    Handle RejectedExecutionException at closing ManagedLedgerClientFactory (#2646)
    
    *Motivation*
    
    Following stack trace is observed in CI.
    
    ```
    Error Message
    java.io.IOException: java.util.concurrent.RejectedExecutionException: Task org.apache.bookkeeper.mledger.util.SafeRun$1@7ec176da
rejected from java.util.concurrent.ThreadPoolExecutor@33d2f9d3[Terminated, pool size = 0,
active threads = 0, queued tasks = 0, completed tasks = 2377]
    Stacktrace
    org.apache.pulsar.broker.PulsarServerException: java.io.IOException: java.util.concurrent.RejectedExecutionException:
Task org.apache.bookkeeper.mledger.util.SafeRun$1@7ec176da rejected from java.util.concurrent.ThreadPoolExecutor@33d2f9d3[Terminated,
pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 2377]
    	at org.apache.pulsar.broker.PulsarService.close(PulsarService.java:298)
    	at org.apache.pulsar.functions.worker.PulsarWorkerAssignmentTest.shutdown(PulsarWorkerAssignmentTest.java:147)
    	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    	at java.lang.reflect.Method.invoke(Method.java:498)
    	at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:124)
    	at org.testng.internal.MethodInvocationHelper.invokeMethodConsideringTimeout(MethodInvocationHelper.java:59)
    	at org.testng.internal.Invoker.invokeConfigurationMethod(Invoker.java:451)
    	at org.testng.internal.Invoker.invokeConfigurations(Invoker.java:222)
    	at org.testng.internal.Invoker.invokeMethod(Invoker.java:634)
    	at org.testng.internal.Invoker.retryFailed(Invoker.java:829)
    	at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:1000)
    	at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:125)
    	at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:109)
    	at org.testng.TestRunner.privateRun(TestRunner.java:648)
    	at org.testng.TestRunner.run(TestRunner.java:505)
    	at org.testng.SuiteRunner.runTest(SuiteRunner.java:455)
    	at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:450)
    	at org.testng.SuiteRunner.privateRun(SuiteRunner.java:415)
    	at org.testng.SuiteRunner.run(SuiteRunner.java:364)
    	at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:52)
    	at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:84)
    	at org.testng.TestNG.runSuitesSequentially(TestNG.java:1187)
    	at org.testng.TestNG.runSuitesLocally(TestNG.java:1116)
    	at org.testng.TestNG.runSuites(TestNG.java:1028)
    	at org.testng.TestNG.run(TestNG.java:996)
    	at org.apache.maven.surefire.testng.TestNGExecutor.run(TestNGExecutor.java:135)
    	at org.apache.maven.surefire.testng.TestNGDirectoryTestSuite.executeSingleClass(TestNGDirectoryTestSuite.java:112)
    	at org.apache.maven.surefire.testng.TestNGDirectoryTestSuite.execute(TestNGDirectoryTestSuite.java:99)
    	at org.apache.maven.surefire.testng.TestNGProvider.invoke(TestNGProvider.java:146)
    	at org.apache.maven.surefire.booter.ForkedBooter.invokeProviderInSameClassLoader(ForkedBooter.java:379)
    	at org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:340)
    	at org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:125)
    	at org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:413)
    Caused by: java.io.IOException: java.util.concurrent.RejectedExecutionException: Task
org.apache.bookkeeper.mledger.util.SafeRun$1@7ec176da rejected from java.util.concurrent.ThreadPoolExecutor@33d2f9d3[Terminated,
pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 2377]
    	at org.apache.pulsar.broker.ManagedLedgerClientFactory.close(ManagedLedgerClientFactory.java:70)
    	at org.apache.pulsar.broker.PulsarService.close(PulsarService.java:225)
    	... 34 more
    Caused by: java.util.concurrent.RejectedExecutionException: Task org.apache.bookkeeper.mledger.util.SafeRun$1@7ec176da
rejected from java.util.concurrent.ThreadPoolExecutor@33d2f9d3[Terminated, pool size = 0,
active threads = 0, queued tasks = 0, completed tasks = 2377]
    	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
    	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
    	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
    	at com.google.common.util.concurrent.ForwardingExecutorService.execute(ForwardingExecutorService.java:99)
    	at org.apache.bookkeeper.common.util.BoundedExecutorService.execute(BoundedExecutorService.java:89)
    	at org.apache.bookkeeper.common.util.OrderedExecutor.executeOrdered(OrderedExecutor.java:291)
    	at org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.lambda$initializeBookKeeper$2(ManagedLedgerImpl.java:376)
    	at org.apache.bookkeeper.client.LedgerCreateOp.createComplete(LedgerCreateOp.java:240)
    	at org.apache.bookkeeper.client.LedgerCreateOp.operationComplete(LedgerCreateOp.java:199)
    	at org.apache.bookkeeper.client.LedgerCreateOp.operationComplete(LedgerCreateOp.java:59)
    	at org.apache.bookkeeper.meta.CleanupLedgerManager.close(CleanupLedgerManager.java:232)
    	at org.apache.bookkeeper.client.BookKeeper.close(BookKeeper.java:1467)
    	at org.apache.pulsar.broker.ManagedLedgerClientFactory.close(ManagedLedgerClientFactory.java:66)
    	... 35 more
    
    ```
    
    The race condition happens when:
    
    when closing bookkeeper client, it will error outs all pending metadata operations.
    those callbacks of those operations will be triggered, and submitted to the scheduler
    in managed ledger factory. but the managed ledger factory has been shutdown before,
    so `RejectedExecutionException` will be thrown there.
    
    we can safely ignore this exception.
    
    *Changes*
    
    Catch RejectedExecutionException and ignore it
---
 .../apache/pulsar/broker/ManagedLedgerClientFactory.java | 14 +++++++++++++-
 .../functions/worker/PulsarWorkerAssignmentTest.java     | 16 ++++++++++------
 2 files changed, 23 insertions(+), 7 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java
index 3189df2..2b82204 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker;
 import java.io.Closeable;
 import java.io.IOException;
 
+import java.util.concurrent.RejectedExecutionException;
 import org.apache.bookkeeper.client.BookKeeper;
 import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
 import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig;
@@ -63,7 +64,18 @@ public class ManagedLedgerClientFactory implements Closeable {
             managedLedgerFactory.shutdown();
             log.info("Closed managed ledger factory");
 
-            bkClient.close();
+            try {
+                bkClient.close();
+            } catch (RejectedExecutionException ree) {
+                // when closing bookkeeper client, it will error outs all pending metadata
operations.
+                // those callbacks of those operations will be triggered, and submitted to
the scheduler
+                // in managed ledger factory. but the managed ledger factory has been shutdown
before,
+                // so `RejectedExecutionException` will be thrown there. we can safely ignore
this exception.
+                //
+                // an alternative solution is to close bookkeeper client before shutting
down managed ledger
+                // factory, however that might be introducing more unknowns.
+                log.warn("Encountered exceptions on closing bookkeeper client", ree);
+            }
             log.info("Closed BookKeeper client");
         } catch (Exception e) {
             log.warn(e.getMessage(), e);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
index fe28a51..8f18259 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarWorkerAssignmentTest.java
@@ -139,13 +139,17 @@ public class PulsarWorkerAssignmentTest {
     }
 
     @AfterMethod
-    void shutdown() throws Exception {
+    void shutdown() {
         log.info("--- Shutting down ---");
-        pulsarClient.close();
-        admin.close();
-        functionsWorkerService.stop();
-        pulsar.close();
-        bkEnsemble.stop();
+        try {
+            pulsarClient.close();
+            admin.close();
+            functionsWorkerService.stop();
+            pulsar.close();
+            bkEnsemble.stop();
+        } catch (Exception e) {
+            log.warn("Encountered errors at shutting down PulsarWorkerAssignmentTest", e);
+        }
     }
 
     private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) {


Mime
View raw message