Skip to content

Reorder the sequence of the bookkeeper server shutdown so that there are no read/ write ops while shutting down the bookie #2888

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.

Sign up for GitHub

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

Merged
Vanlightly merged 8 commits into from
Dec 1, 2021
Merged

Reorder the sequence of the bookkeeper server shutdown so that there are no read/ write ops while shutting down the bookie #2888

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
9f00279
Reorder the sequence of the bookkeeper server shutdown so that there …
pradeepbn Nov 10, 2021
7fec12a
Fix the force shutdown on the requestProcess; Check if the channel is…
pradeepbn Nov 12, 2021
66b2513
Revert the Dockefile changes
pradeepbn Nov 16, 2021
5a29761
Add channel.isActive() mock
pradeepbn Nov 18, 2021
f8c2525
Revert nettyServer shutdown sequence
pradeepbn Nov 18, 2021
b7908b9
Added a log statement at the close of request processor for tracking …
pradeepbn Nov 18, 2021
e6b5501
Adding logs when the netty channel is inactive to bypass the writeAnd…
pradeepbn Nov 23, 2021
15abe17
Adding the channel information to the debug logs while the channel is…
pradeepbn Nov 30, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieRequestProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -264,13 +264,15 @@ int maxReadsInProgressCount() {

@Override
public void close() {
LOG.info("Closing RequestProcessor");
Comment thread
Vanlightly marked this conversation as resolved.
shutdownExecutor(writeThreadPool);
shutdownExecutor(readThreadPool);
if (serverCfg.getNumLongPollWorkerThreads() > 0 || readThreadPool == null) {
shutdownExecutor(longPollThreadPool);
}
shutdownExecutor(highPriorityThreadPool);
requestTimer.stop();
LOG.info("Closed RequestProcessor");
}

private OrderedExecutor createExecutor(
Expand All @@ -295,6 +297,7 @@ private OrderedExecutor createExecutor(
private void shutdownExecutor(OrderedExecutor service) {
if (null != service) {
service.shutdown();
service.forceShutdown(10, TimeUnit.SECONDS);
}
}

Expand Down
2 changes: 1 addition & 1 deletion bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,8 +220,8 @@ public synchronized void shutdown() {
if (!running) {
return;
}
exitCode = bookie.shutdown();
this.requestProcessor.close();
exitCode = bookie.shutdown();
running = false;
}

Expand Down
7 changes: 6 additions & 1 deletion bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PacketProcessorBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,12 @@ protected boolean isVersionCompatible() {
}

protected void sendResponse(int rc, Object response, OpStatsLogger statsLogger) {
channel.writeAndFlush(response, channel.voidPromise());
if (channel.isActive()) {

@eolivelli eolivelli Nov 22, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what about adding a "ELSE" branch with a log message ?

@Vanlightly Vanlightly Nov 22, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently what happens when the BookieServer is shutdown is that for each in-progress op, when the response is sent it will fail because the channel is closed, and that will get logged as a metric:

if (!future.isSuccess()) {
    requestProcessor.getRequestStats().getChannelWriteStats()
                        .registerFailedEvent(writeElapsedNanos, TimeUnit.NANOSECONDS);
} else {
    requestProcessor.getRequestStats().getChannelWriteStats()
                        .registerSuccessfulEvent(writeElapsedNanos, TimeUnit.NANOSECONDS);
 }

So either we add an ELSE branch with the same registerFailedEvent call, or we don't use an IF at all and allow it to use the existing logic flow.

@pradeepbn what was the reason for avoiding channel.writeAndFlush?

@Vanlightly Vanlightly Nov 22, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd say this change is not strictly related to shutdown reordering, so I think removing these channel checks can be removed from this PR.

@pradeepbn pradeepbn Nov 23, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Vanlightly

bookie1_1 | 02:15:08,926 ERROR Failed to submit a listener notification task. Event loop shut down? bookie1_1 | java.util.concurrent.RejectedExecutionException: event executor terminated bookie1_1 | at io.netty.util.concurrent.SingleThreadEventExecutor.reject(SingleThreadEventExecutor.java:923) ~[netty-common-4.1.68.Final.jar:4.1.68.Final] bookie1_1 | at io.netty.util.concurrent.SingleThreadEventExecutor.offerTask(SingleThreadEventExecutor.java:350) ~[netty-common-4.1.68.Final.jar:4.1.68.Final] bookie1_1 | at io.netty.util.concurrent.SingleThreadEventExecutor.addTask(SingleThreadEventExecutor.java:343) ~[netty-common-4.1.68.Final.jar:4.1.68.Final] bookie1_1 | at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:825) ~[netty-common-4.1.68.Final.jar:4.1.68.Final] bookie1_1 | at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:815) ~[netty-common-4.1.68.Final.jar:4.1.68.Final] bookie1_1 | at io.netty.util.concurrent.DefaultPromise.safeExecute(DefaultPromise.java:841) [netty-common-4.1.68.Final.jar:4.1.68.Final] bookie1_1 | at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:499) [netty-common-4.1.68.Final.jar:4.1.68.Final] bookie1_1 | at io.netty.util.concurrent.DefaultPromise.addListener(DefaultPromise.java:184) [netty-common-4.1.68.Final.jar:4.1.68.Final] bookie1_1 | at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:95) [netty-transport-4.1.68.Final.jar:4.1.68.Final] bookie1_1 | at io.netty.channel.DefaultChannelPromise.addListener(DefaultChannelPromise.java:30) [netty-transport-4.1.68.Final.jar:4.1.68.Final] bookie1_1 | at org.apache.bookkeeper.proto.PacketProcessorBaseV3.sendResponse(PacketProcessorBaseV3.java:91) [bookkeeper-server.jar:?] bookie1_1 | at org.apache.bookkeeper.proto.WriteEntryProcessorV3.sendResponse(WriteEntryProcessorV3.java:180) [bookkeeper-server.jar:?] bookie1_1 | at org.apache.bookkeeper.proto.WriteEntryProcessorV3$1.writeComplete(WriteEntryProcessorV3.java:108) [bookkeeper-server.jar:?] bookie1_1 | at org.apache.bookkeeper.bookie.Journal$QueueEntry.run(Journal.java:335) [bookkeeper-server.jar:?] bookie1_1 | at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?] bookie1_1 | at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?] bookie1_1 | at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-common-4.1.68.Final.jar:4.1.68.Final] bookie1_1 | at java.lang.Thread.run(Thread.java:829) [?:?]

If we do not check for isActive(), we will get this exception on channel.writeAndFlush at the time of shutdown. This happens because of nettyserver shutdown before the request processor.

@Vanlightly Vanlightly Nov 23, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a controlled exception within Netty but it will pollute the log during shutdown. So I think that's its worth keeping the IF statement. What do you think @eolivelli?

@eolivelli eolivelli Nov 23, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok, let's keep the original "if"

we could add an "else" branch with a DEBUG log that says that we are skipping the write, this would help in debugging tests failures probably one day

@pradeepbn pradeepbn Nov 23, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added the else statement with the logs. CC: @eolivelli @Vanlightly

channel.writeAndFlush(response, channel.voidPromise());
} else {
LOGGER.debug("Netty channel {} is inactive, "
+ "hence bypassing netty channel writeAndFlush during sendResponse", channel);
}
if (BookieProtocol.EOK == rc) {
statsLogger.registerSuccessfulEvent(MathUtils.elapsedNanos(enqueueNanos), TimeUnit.NANOSECONDS);
} else {
Expand Down
40 changes: 22 additions & 18 deletions bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PacketProcessorBaseV3.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,25 +87,29 @@ protected void sendResponse(StatusCode code, Object response, OpStatsLogger stat
requestProcessor.invalidateBlacklist(channel);
}
}

channel.writeAndFlush(response).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
long writeElapsedNanos = MathUtils.elapsedNanos(writeNanos);
if (!future.isSuccess()) {
requestProcessor.getRequestStats().getChannelWriteStats()
.registerFailedEvent(writeElapsedNanos, TimeUnit.NANOSECONDS);
} else {
requestProcessor.getRequestStats().getChannelWriteStats()
.registerSuccessfulEvent(writeElapsedNanos, TimeUnit.NANOSECONDS);
}
if (StatusCode.EOK == code) {
statsLogger.registerSuccessfulEvent(MathUtils.elapsedNanos(enqueueNanos), TimeUnit.NANOSECONDS);
} else {
statsLogger.registerFailedEvent(MathUtils.elapsedNanos(enqueueNanos), TimeUnit.NANOSECONDS);
if (channel.isActive()) {
Comment thread
Vanlightly marked this conversation as resolved.

@eolivelli eolivelli Nov 22, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what about adding a "ELSE" branch with a log message ?

@pradeepbn pradeepbn Nov 23, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

channel.writeAndFlush(response).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
long writeElapsedNanos = MathUtils.elapsedNanos(writeNanos);
if (!future.isSuccess()) {
requestProcessor.getRequestStats().getChannelWriteStats()
.registerFailedEvent(writeElapsedNanos, TimeUnit.NANOSECONDS);
} else {
requestProcessor.getRequestStats().getChannelWriteStats()
.registerSuccessfulEvent(writeElapsedNanos, TimeUnit.NANOSECONDS);
}
if (StatusCode.EOK == code) {
statsLogger.registerSuccessfulEvent(MathUtils.elapsedNanos(enqueueNanos), TimeUnit.NANOSECONDS);
} else {
statsLogger.registerFailedEvent(MathUtils.elapsedNanos(enqueueNanos), TimeUnit.NANOSECONDS);
}
}
}
});
});
} else {
LOGGER.debug("Netty channel {} is inactive, "
+ "hence bypassing netty channel writeAndFlush during sendResponse", channel);
}
}

protected boolean isVersionCompatible() {
Expand Down
1 change: 1 addition & 0 deletions bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/ForceLedgerProcessorV3Test.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,7 @@ public void setup() {
when(requestProcessor.getBookie()).thenReturn(bookie);
when(requestProcessor.getWaitTimeoutOnBackpressureMillis()).thenReturn(-1L);
when(requestProcessor.getRequestStats()).thenReturn(new RequestStats(NullStatsLogger.INSTANCE));
when(channel.isActive()).thenReturn(true);
processor = new ForceLedgerProcessorV3(
request,
channel,
Expand Down
1 change: 1 addition & 0 deletions bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/WriteEntryProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ public void setup() {
requestProcessor = mock(BookieRequestProcessor.class);
when(requestProcessor.getBookie()).thenReturn(bookie);
when(requestProcessor.getRequestStats()).thenReturn(new RequestStats(NullStatsLogger.INSTANCE));
when(channel.isActive()).thenReturn(true);
processor = WriteEntryProcessor.create(
request,
channel,
Expand Down
1 change: 1 addition & 0 deletions bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3Test.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ public void setup() {
when(requestProcessor.getBookie()).thenReturn(bookie);
when(requestProcessor.getWaitTimeoutOnBackpressureMillis()).thenReturn(-1L);
when(requestProcessor.getRequestStats()).thenReturn(new RequestStats(NullStatsLogger.INSTANCE));
when(channel.isActive()).thenReturn(true);
processor = new WriteEntryProcessorV3(
request,
channel,
Expand Down