-
Notifications
You must be signed in to change notification settings - Fork 896
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[fix] Clear channel when channelInactive #3966
[fix] Clear channel when channelInactive #3966
Conversation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good catch.
I see that in all the other places in which we set state = ConnectionState.DISCONNECT
we also set channel to null.
+1
rerun failure checks |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Nice catch.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Nice catch!
### Motivation I am stopping a pulsar broker, and I observed a lot of the following logs in the broker log: ``` 17:48:35.894 [shutdown-thread-43-1] INFO org.apache.bookkeeper.proto.PerChannelBookieClient - Closing the per channel bookie client for 10.184.xx.xx:3181 17:48:35.894 [shutdown-thread-43-1] ERROR io.netty.util.concurrent.DefaultPromise.rejectedExecution - Failed to submit a listener notification task. Event loop shut down? java.util.concurrent.RejectedExecutionException: event executor terminated at io.netty.util.concurrent.SingleThreadEventExecutor.reject(SingleThreadEventExecutor.java:923) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.offerTask(SingleThreadEventExecutor.java:350) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.addTask(SingleThreadEventExecutor.java:343) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:825) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:815) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.safeExecute(DefaultPromise.java:842) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:499) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.addListener(DefaultPromise.java:184) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:95) ~[io.netty-netty-transport-4.1.66.Final.jar:4.1.66.Final] at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:30) ~[io.netty-netty-transport-4.1.66.Final.jar:4.1.66.Final] at org.apache.bookkeeper.proto.PerChannelBookieClient.closeChannel(PerChannelBookieClient.java:1090) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.PerChannelBookieClient.closeInternal(PerChannelBookieClient.java:1079) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.PerChannelBookieClient.close(PerChannelBookieClient.java:1063) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.DefaultPerChannelBookieClientPool.close(DefaultPerChannelBookieClientPool.java:157) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.BookieClientImpl.close(BookieClientImpl.java:587) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.client.BookKeeper.close(BookKeeper.java:1435) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.pulsar.broker.ManagedLedgerClientFactory.close(ManagedLedgerClientFactory.java:142) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at org.apache.pulsar.broker.PulsarService.closeAsync(PulsarService.java:417) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at org.apache.pulsar.broker.MessagingServiceShutdownHook.lambda$run$1(MessagingServiceShutdownHook.java:62) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_131] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_131] at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at java.lang.Thread.run(Thread.java:748) [?:1.8.0_131] ``` When I stopped the broker, there were more than 30,000 lines in such a log, which seriously polluted my normal log of the broker. <img width="1180" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/0ec3c423-14f1-41c4-85c3-d978ece15044"> According to my next investigation, 1. When the broker stops, it will close the ManagedLedgerClientFactory. It will close all bookkeeper connections https://github.com/apache/bookkeeper/blob/master/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/DefaultPerChannelBookieClientPool.java#L154 ```java @OverRide public void close(boolean wait) { for (int i = 0; i < clients.length; i++) { clients[i].close(wait); if (clients != clientsV3Enforced) { clientsV3Enforced[i].close(wait); } } } private void closeInternal(boolean permanent, boolean wait) { Channel toClose = null; synchronized (this) { if (permanent) { state = ConnectionState.CLOSED; } else if (state != ConnectionState.CLOSED) { state = ConnectionState.DISCONNECTED; } toClose = channel; channel = null; makeWritable(); } if (toClose != null) { <== a. toClose not null ChannelFuture cf = closeChannel(toClose); if (wait) { cf.awaitUninterruptibly(); } } } private ChannelFuture closeChannel(Channel c) { if (LOG.isDebugEnabled()) { LOG.debug("Closing channel {}", c); } return c.close().addListener(x -> makeWritable()); <== b. here if channel is already inactive, add the listener will throw exception } ``` 2. When the bookie server first disconnects the client, the client will trigger channelInactive ```java public void channelInactive(ChannelHandlerContext ctx) throws Exception { LOG.info("Disconnected from bookie channel {}", ctx.channel()); if (ctx.channel() != null) { closeChannel(ctx.channel()); <== c. here channel is closed if (ctx.channel().pipeline().get(SslHandler.class) != null) { activeTlsChannelCounter.dec(); } else { activeNonTlsChannelCounter.dec(); } } errorOutOutstandingEntries(BKException.Code.BookieHandleNotAvailableException); errorOutPendingOps(BKException.Code.BookieHandleNotAvailableException); synchronized (this) { if (this.channel == ctx.channel() && state != ConnectionState.CLOSED) { state = ConnectionState.DISCONNECTED; } } // we don't want to reconnect right away. If someone sends a request to // this address, we will reconnect. } ``` If step c happens before b, then b throws the above exception. step c <img width="1400" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/4a33855c-5682-4812-ae0e-1a02972ef502"> <img width="1407" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/1c99bccb-b2bd-4cfb-ac4b-ee10b2d236b0"> step a and b <img width="1403" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/fec372c5-f97b-47e7-b5d7-ebbaa6edf682"> So when channelInactive, we should reset the channel to null after closed the channel. Here, when the channel is null, we will reconnect after getting the client, which is consistent with the current behavior. ![image](https://github.com/apache/bookkeeper/assets/35599757/d1178697-27bd-44a5-acd1-33ccec99e7a4) (cherry picked from commit 808b909)
### Motivation I am stopping a pulsar broker, and I observed a lot of the following logs in the broker log: ``` 17:48:35.894 [shutdown-thread-43-1] INFO org.apache.bookkeeper.proto.PerChannelBookieClient - Closing the per channel bookie client for 10.184.xx.xx:3181 17:48:35.894 [shutdown-thread-43-1] ERROR io.netty.util.concurrent.DefaultPromise.rejectedExecution - Failed to submit a listener notification task. Event loop shut down? java.util.concurrent.RejectedExecutionException: event executor terminated at io.netty.util.concurrent.SingleThreadEventExecutor.reject(SingleThreadEventExecutor.java:923) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.offerTask(SingleThreadEventExecutor.java:350) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.addTask(SingleThreadEventExecutor.java:343) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:825) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:815) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.safeExecute(DefaultPromise.java:842) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:499) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.addListener(DefaultPromise.java:184) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:95) ~[io.netty-netty-transport-4.1.66.Final.jar:4.1.66.Final] at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:30) ~[io.netty-netty-transport-4.1.66.Final.jar:4.1.66.Final] at org.apache.bookkeeper.proto.PerChannelBookieClient.closeChannel(PerChannelBookieClient.java:1090) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.PerChannelBookieClient.closeInternal(PerChannelBookieClient.java:1079) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.PerChannelBookieClient.close(PerChannelBookieClient.java:1063) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.DefaultPerChannelBookieClientPool.close(DefaultPerChannelBookieClientPool.java:157) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.BookieClientImpl.close(BookieClientImpl.java:587) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.client.BookKeeper.close(BookKeeper.java:1435) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.pulsar.broker.ManagedLedgerClientFactory.close(ManagedLedgerClientFactory.java:142) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at org.apache.pulsar.broker.PulsarService.closeAsync(PulsarService.java:417) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at org.apache.pulsar.broker.MessagingServiceShutdownHook.lambda$run$1(MessagingServiceShutdownHook.java:62) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_131] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_131] at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at java.lang.Thread.run(Thread.java:748) [?:1.8.0_131] ``` When I stopped the broker, there were more than 30,000 lines in such a log, which seriously polluted my normal log of the broker. <img width="1180" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/0ec3c423-14f1-41c4-85c3-d978ece15044"> According to my next investigation, 1. When the broker stops, it will close the ManagedLedgerClientFactory. It will close all bookkeeper connections https://github.com/apache/bookkeeper/blob/master/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/DefaultPerChannelBookieClientPool.java#L154 ```java @OverRide public void close(boolean wait) { for (int i = 0; i < clients.length; i++) { clients[i].close(wait); if (clients != clientsV3Enforced) { clientsV3Enforced[i].close(wait); } } } private void closeInternal(boolean permanent, boolean wait) { Channel toClose = null; synchronized (this) { if (permanent) { state = ConnectionState.CLOSED; } else if (state != ConnectionState.CLOSED) { state = ConnectionState.DISCONNECTED; } toClose = channel; channel = null; makeWritable(); } if (toClose != null) { <== a. toClose not null ChannelFuture cf = closeChannel(toClose); if (wait) { cf.awaitUninterruptibly(); } } } private ChannelFuture closeChannel(Channel c) { if (LOG.isDebugEnabled()) { LOG.debug("Closing channel {}", c); } return c.close().addListener(x -> makeWritable()); <== b. here if channel is already inactive, add the listener will throw exception } ``` 2. When the bookie server first disconnects the client, the client will trigger channelInactive ```java public void channelInactive(ChannelHandlerContext ctx) throws Exception { LOG.info("Disconnected from bookie channel {}", ctx.channel()); if (ctx.channel() != null) { closeChannel(ctx.channel()); <== c. here channel is closed if (ctx.channel().pipeline().get(SslHandler.class) != null) { activeTlsChannelCounter.dec(); } else { activeNonTlsChannelCounter.dec(); } } errorOutOutstandingEntries(BKException.Code.BookieHandleNotAvailableException); errorOutPendingOps(BKException.Code.BookieHandleNotAvailableException); synchronized (this) { if (this.channel == ctx.channel() && state != ConnectionState.CLOSED) { state = ConnectionState.DISCONNECTED; } } // we don't want to reconnect right away. If someone sends a request to // this address, we will reconnect. } ``` If step c happens before b, then b throws the above exception. step c <img width="1400" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/4a33855c-5682-4812-ae0e-1a02972ef502"> <img width="1407" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/1c99bccb-b2bd-4cfb-ac4b-ee10b2d236b0"> step a and b <img width="1403" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/fec372c5-f97b-47e7-b5d7-ebbaa6edf682"> So when channelInactive, we should reset the channel to null after closed the channel. Here, when the channel is null, we will reconnect after getting the client, which is consistent with the current behavior. ![image](https://github.com/apache/bookkeeper/assets/35599757/d1178697-27bd-44a5-acd1-33ccec99e7a4) (cherry picked from commit 808b909)
### Motivation I am stopping a pulsar broker, and I observed a lot of the following logs in the broker log: ``` 17:48:35.894 [shutdown-thread-43-1] INFO org.apache.bookkeeper.proto.PerChannelBookieClient - Closing the per channel bookie client for 10.184.xx.xx:3181 17:48:35.894 [shutdown-thread-43-1] ERROR io.netty.util.concurrent.DefaultPromise.rejectedExecution - Failed to submit a listener notification task. Event loop shut down? java.util.concurrent.RejectedExecutionException: event executor terminated at io.netty.util.concurrent.SingleThreadEventExecutor.reject(SingleThreadEventExecutor.java:923) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.offerTask(SingleThreadEventExecutor.java:350) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.addTask(SingleThreadEventExecutor.java:343) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:825) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:815) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.safeExecute(DefaultPromise.java:842) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:499) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.addListener(DefaultPromise.java:184) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:95) ~[io.netty-netty-transport-4.1.66.Final.jar:4.1.66.Final] at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:30) ~[io.netty-netty-transport-4.1.66.Final.jar:4.1.66.Final] at org.apache.bookkeeper.proto.PerChannelBookieClient.closeChannel(PerChannelBookieClient.java:1090) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.PerChannelBookieClient.closeInternal(PerChannelBookieClient.java:1079) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.PerChannelBookieClient.close(PerChannelBookieClient.java:1063) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.DefaultPerChannelBookieClientPool.close(DefaultPerChannelBookieClientPool.java:157) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.BookieClientImpl.close(BookieClientImpl.java:587) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.client.BookKeeper.close(BookKeeper.java:1435) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.pulsar.broker.ManagedLedgerClientFactory.close(ManagedLedgerClientFactory.java:142) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at org.apache.pulsar.broker.PulsarService.closeAsync(PulsarService.java:417) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at org.apache.pulsar.broker.MessagingServiceShutdownHook.lambda$run$1(MessagingServiceShutdownHook.java:62) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_131] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_131] at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at java.lang.Thread.run(Thread.java:748) [?:1.8.0_131] ``` When I stopped the broker, there were more than 30,000 lines in such a log, which seriously polluted my normal log of the broker. <img width="1180" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/0ec3c423-14f1-41c4-85c3-d978ece15044"> According to my next investigation, 1. When the broker stops, it will close the ManagedLedgerClientFactory. It will close all bookkeeper connections https://github.com/apache/bookkeeper/blob/master/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/DefaultPerChannelBookieClientPool.java#L154 ```java @OverRide public void close(boolean wait) { for (int i = 0; i < clients.length; i++) { clients[i].close(wait); if (clients != clientsV3Enforced) { clientsV3Enforced[i].close(wait); } } } private void closeInternal(boolean permanent, boolean wait) { Channel toClose = null; synchronized (this) { if (permanent) { state = ConnectionState.CLOSED; } else if (state != ConnectionState.CLOSED) { state = ConnectionState.DISCONNECTED; } toClose = channel; channel = null; makeWritable(); } if (toClose != null) { <== a. toClose not null ChannelFuture cf = closeChannel(toClose); if (wait) { cf.awaitUninterruptibly(); } } } private ChannelFuture closeChannel(Channel c) { if (LOG.isDebugEnabled()) { LOG.debug("Closing channel {}", c); } return c.close().addListener(x -> makeWritable()); <== b. here if channel is already inactive, add the listener will throw exception } ``` 2. When the bookie server first disconnects the client, the client will trigger channelInactive ```java public void channelInactive(ChannelHandlerContext ctx) throws Exception { LOG.info("Disconnected from bookie channel {}", ctx.channel()); if (ctx.channel() != null) { closeChannel(ctx.channel()); <== c. here channel is closed if (ctx.channel().pipeline().get(SslHandler.class) != null) { activeTlsChannelCounter.dec(); } else { activeNonTlsChannelCounter.dec(); } } errorOutOutstandingEntries(BKException.Code.BookieHandleNotAvailableException); errorOutPendingOps(BKException.Code.BookieHandleNotAvailableException); synchronized (this) { if (this.channel == ctx.channel() && state != ConnectionState.CLOSED) { state = ConnectionState.DISCONNECTED; } } // we don't want to reconnect right away. If someone sends a request to // this address, we will reconnect. } ``` If step c happens before b, then b throws the above exception. step c <img width="1400" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/4a33855c-5682-4812-ae0e-1a02972ef502"> <img width="1407" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/1c99bccb-b2bd-4cfb-ac4b-ee10b2d236b0"> step a and b <img width="1403" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/fec372c5-f97b-47e7-b5d7-ebbaa6edf682"> So when channelInactive, we should reset the channel to null after closed the channel. Here, when the channel is null, we will reconnect after getting the client, which is consistent with the current behavior. ![image](https://github.com/apache/bookkeeper/assets/35599757/d1178697-27bd-44a5-acd1-33ccec99e7a4) (cherry picked from commit 808b909)
### Motivation I am stopping a pulsar broker, and I observed a lot of the following logs in the broker log: ``` 17:48:35.894 [shutdown-thread-43-1] INFO org.apache.bookkeeper.proto.PerChannelBookieClient - Closing the per channel bookie client for 10.184.xx.xx:3181 17:48:35.894 [shutdown-thread-43-1] ERROR io.netty.util.concurrent.DefaultPromise.rejectedExecution - Failed to submit a listener notification task. Event loop shut down? java.util.concurrent.RejectedExecutionException: event executor terminated at io.netty.util.concurrent.SingleThreadEventExecutor.reject(SingleThreadEventExecutor.java:923) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.offerTask(SingleThreadEventExecutor.java:350) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.addTask(SingleThreadEventExecutor.java:343) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:825) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:815) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.safeExecute(DefaultPromise.java:842) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:499) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.util.concurrent.DefaultPromise.addListener(DefaultPromise.java:184) ~[io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:95) ~[io.netty-netty-transport-4.1.66.Final.jar:4.1.66.Final] at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:30) ~[io.netty-netty-transport-4.1.66.Final.jar:4.1.66.Final] at org.apache.bookkeeper.proto.PerChannelBookieClient.closeChannel(PerChannelBookieClient.java:1090) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.PerChannelBookieClient.closeInternal(PerChannelBookieClient.java:1079) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.PerChannelBookieClient.close(PerChannelBookieClient.java:1063) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.DefaultPerChannelBookieClientPool.close(DefaultPerChannelBookieClientPool.java:157) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.proto.BookieClientImpl.close(BookieClientImpl.java:587) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.bookkeeper.client.BookKeeper.close(BookKeeper.java:1435) ~[org.apache.bookkeeper-bookkeeper-server-4.14.2.jar:4.14.2] at org.apache.pulsar.broker.ManagedLedgerClientFactory.close(ManagedLedgerClientFactory.java:142) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at org.apache.pulsar.broker.PulsarService.closeAsync(PulsarService.java:417) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at org.apache.pulsar.broker.MessagingServiceShutdownHook.lambda$run$1(MessagingServiceShutdownHook.java:62) ~[org.apache.pulsar-pulsar-broker-2.8.1.jar:2.8.1] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_131] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_131] at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.66.Final.jar:4.1.66.Final] at java.lang.Thread.run(Thread.java:748) [?:1.8.0_131] ``` When I stopped the broker, there were more than 30,000 lines in such a log, which seriously polluted my normal log of the broker. <img width="1180" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/0ec3c423-14f1-41c4-85c3-d978ece15044"> According to my next investigation, 1. When the broker stops, it will close the ManagedLedgerClientFactory. It will close all bookkeeper connections https://github.com/apache/bookkeeper/blob/master/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/DefaultPerChannelBookieClientPool.java#L154 ```java @OverRide public void close(boolean wait) { for (int i = 0; i < clients.length; i++) { clients[i].close(wait); if (clients != clientsV3Enforced) { clientsV3Enforced[i].close(wait); } } } private void closeInternal(boolean permanent, boolean wait) { Channel toClose = null; synchronized (this) { if (permanent) { state = ConnectionState.CLOSED; } else if (state != ConnectionState.CLOSED) { state = ConnectionState.DISCONNECTED; } toClose = channel; channel = null; makeWritable(); } if (toClose != null) { <== a. toClose not null ChannelFuture cf = closeChannel(toClose); if (wait) { cf.awaitUninterruptibly(); } } } private ChannelFuture closeChannel(Channel c) { if (LOG.isDebugEnabled()) { LOG.debug("Closing channel {}", c); } return c.close().addListener(x -> makeWritable()); <== b. here if channel is already inactive, add the listener will throw exception } ``` 2. When the bookie server first disconnects the client, the client will trigger channelInactive ```java public void channelInactive(ChannelHandlerContext ctx) throws Exception { LOG.info("Disconnected from bookie channel {}", ctx.channel()); if (ctx.channel() != null) { closeChannel(ctx.channel()); <== c. here channel is closed if (ctx.channel().pipeline().get(SslHandler.class) != null) { activeTlsChannelCounter.dec(); } else { activeNonTlsChannelCounter.dec(); } } errorOutOutstandingEntries(BKException.Code.BookieHandleNotAvailableException); errorOutPendingOps(BKException.Code.BookieHandleNotAvailableException); synchronized (this) { if (this.channel == ctx.channel() && state != ConnectionState.CLOSED) { state = ConnectionState.DISCONNECTED; } } // we don't want to reconnect right away. If someone sends a request to // this address, we will reconnect. } ``` If step c happens before b, then b throws the above exception. step c <img width="1400" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/4a33855c-5682-4812-ae0e-1a02972ef502"> <img width="1407" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/1c99bccb-b2bd-4cfb-ac4b-ee10b2d236b0"> step a and b <img width="1403" alt="image" src="https://github.com/apache/bookkeeper/assets/35599757/fec372c5-f97b-47e7-b5d7-ebbaa6edf682"> So when channelInactive, we should reset the channel to null after closed the channel. Here, when the channel is null, we will reconnect after getting the client, which is consistent with the current behavior. ![image](https://github.com/apache/bookkeeper/assets/35599757/d1178697-27bd-44a5-acd1-33ccec99e7a4)
Motivation
I am stopping a pulsar broker, and I observed a lot of the following logs in the broker log:
When I stopped the broker, there were more than 30,000 lines in such a log, which seriously polluted my normal log of the broker.
![image](https://private-user-images.githubusercontent.com/35599757/239962727-0ec3c423-14f1-41c4-85c3-d978ece15044.png?jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3MjQzMTQ1MTQsIm5iZiI6MTcyNDMxNDIxNCwicGF0aCI6Ii8zNTU5OTc1Ny8yMzk5NjI3MjctMGVjM2M0MjMtMTRmMS00MWM0LTg1YzMtZDk3OGVjZTE1MDQ0LnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNDA4MjIlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjQwODIyVDA4MTAxNFomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTJlZmJlOWY2ZDhmOGI1MzJlYTcxNGYwMWMzZjFhZjMwODhjOGQzYWJmNDRmODIxMGU3NGEwNTZmYTZmYTJhNWUmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JmFjdG9yX2lkPTAma2V5X2lkPTAmcmVwb19pZD0wIn0._hpAAWEj_OZRpgyY5XU7doOXDfv143L7YguDNNx40W0)
According to my next investigation,
https://github.com/apache/bookkeeper/blob/master/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/DefaultPerChannelBookieClientPool.java#L154
If step c happens before b, then b throws the above exception.
step c
![image](https://private-user-images.githubusercontent.com/35599757/239966294-4a33855c-5682-4812-ae0e-1a02972ef502.png?jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3MjQzMTQ1MTQsIm5iZiI6MTcyNDMxNDIxNCwicGF0aCI6Ii8zNTU5OTc1Ny8yMzk5NjYyOTQtNGEzMzg1NWMtNTY4Mi00ODEyLWFlMGUtMWEwMjk3MmVmNTAyLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNDA4MjIlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjQwODIyVDA4MTAxNFomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPWY0MmY1OGI2OTg0Mjk2ZjQ2ZjhhOTI0YmIzYWJiMDQyYmY5MjY0ODRhMGJjZjJjZWZkMDRhZmI0MjZkMTBlNjUmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JmFjdG9yX2lkPTAma2V5X2lkPTAmcmVwb19pZD0wIn0.gl7behKUqVn3iRK9eKAHr94dmaVzRob8MQpH5rpLJwg)
![image](https://private-user-images.githubusercontent.com/35599757/239966437-1c99bccb-b2bd-4cfb-ac4b-ee10b2d236b0.png?jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3MjQzMTQ1MTQsIm5iZiI6MTcyNDMxNDIxNCwicGF0aCI6Ii8zNTU5OTc1Ny8yMzk5NjY0MzctMWM5OWJjY2ItYjJiZC00Y2ZiLWFjNGItZWUxMGIyZDIzNmIwLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNDA4MjIlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjQwODIyVDA4MTAxNFomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTViY2RiOGViYmMxOTc4MjQwYWQ4MTQ4NWM2NmU2MjVkOTY2OTFmNGY4YTM0OTM3Y2IzNTU2ZDVjNTNmZjI3YmImWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JmFjdG9yX2lkPTAma2V5X2lkPTAmcmVwb19pZD0wIn0.nnhYQ_I6B9ND-XZoSxvoHvdqKS3siPyBiGSbhq2jtbM)
step a and b
![image](https://private-user-images.githubusercontent.com/35599757/239966599-fec372c5-f97b-47e7-b5d7-ebbaa6edf682.png?jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3MjQzMTQ1MTQsIm5iZiI6MTcyNDMxNDIxNCwicGF0aCI6Ii8zNTU5OTc1Ny8yMzk5NjY1OTktZmVjMzcyYzUtZjk3Yi00N2U3LWI1ZDctZWJiYWE2ZWRmNjgyLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNDA4MjIlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjQwODIyVDA4MTAxNFomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTY0NzhjM2M3MWFkM2ZjZDU4ODg3ZDEyZWU3OTBiNDFkZTQ1MzJiMTg1NzUyNzg1MDA5Yzk2NTBiNDI4ZTI4NmYmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JmFjdG9yX2lkPTAma2V5X2lkPTAmcmVwb19pZD0wIn0.yUaVLwFlr1ACdjfABm7Oq0y7Y91oSOQJmRm8yQ25YnE)
So when channelInactive, we should reset the channel to null after closed the channel.
![image](https://private-user-images.githubusercontent.com/35599757/239967210-d1178697-27bd-44a5-acd1-33ccec99e7a4.png?jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3MjQzMTQ1MTQsIm5iZiI6MTcyNDMxNDIxNCwicGF0aCI6Ii8zNTU5OTc1Ny8yMzk5NjcyMTAtZDExNzg2OTctMjdiZC00NGE1LWFjZDEtMzNjY2VjOTllN2E0LnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNDA4MjIlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjQwODIyVDA4MTAxNFomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTNlMGQ2ZjFlZjVjNzdhNzRjOGQ1ZDM3NjljNzY2OGM5MDk4MWQ0NTBjNTE1M2MxNWJhYzYwODJkMzZjNzVmOWImWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JmFjdG9yX2lkPTAma2V5X2lkPTAmcmVwb19pZD0wIn0.2XcH96Sn7MHsfAVNF7IW7iEc_ngnqyH_FoQvaC3-n0k)
Here, when the channel is null, we will reconnect after getting the client, which is consistent with the current behavior.