Skip to content

[BEAM-14157] Don't call requestObserver.onNext on closed windmill streams #17162

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
TheNeuralBit merged 1 commit into from
Mar 24, 2022
Merged

[BEAM-14157] Don't call requestObserver.onNext on closed windmill streams #17162

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
6 changes: 5 additions & 1 deletion ...worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -848,7 +848,11 @@ public void run() {
new TimerTask() {
@Override
public void run() {
refreshActiveWork();
try {
refreshActiveWork();
} catch (RuntimeException e) {
LOG.warn("Failed to refresh active work: ", e);
Copy link
Member

@TheNeuralBit TheNeuralBit Mar 24, 2022

Choose a reason for hiding this comment

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

Sorry I'm not that familiar with this code, could you help me understand the change?

I'm assuming that this refreshActiveWork() call ultimately calls GrpcWindmiillServier.send(), so we're catching the error here and logging it. This is an improvement because previously that situation caused stalled GetWork streams?

Are there other places where we should be catching the IllegalStateException?

Copy link
Contributor Author

@arunpandianp arunpandianp Mar 24, 2022

Choose a reason for hiding this comment

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

Yes refreshActiveWork ultimately calls GrpcWindmiillServier.send() which could throw. This catch is to supress the exception, refreshActiveWork will get called by the timer again.

Are there other places where we should be catching the IllegalStateException?

Other places where GrpcWindmiillServier.send is called are under try catch already.

This is an improvement because previously that situation caused stalled GetWork streams?

This catch is not directly related to the improvement.

The improvement is that we are now not calling requestObserver.onNext if clientClosed is true, which prevents streams stalling. The new behavior aligns with the grpc-java doc which says don't call onNext on a closed stream https://github.com/grpc/grpc-java/blob/master/stub/src/main/java/io/grpc/stub/StreamObserver.java#L62

Copy link
Member

@TheNeuralBit TheNeuralBit Mar 24, 2022

Choose a reason for hiding this comment

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

Got it, thank you!

}
}
},
options.getActiveWorkRefreshPeriodMillis(),
Expand Down
3 changes: 3 additions & 0 deletions ...er/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/GrpcWindmillServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -663,6 +663,9 @@ protected AbstractWindmillStream(
protected final void send(RequestT request) {
lastSendTimeMs.set(Instant.now().getMillis());
synchronized (this) {
if (clientClosed.get()) {
throw new IllegalStateException("Send called on a client closed stream.");
}
requestObserver.onNext(request);
}
}
Expand Down