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

arunpandianp · 2022-03-23T03:46:22Z

requestObserver.onNext() which is called by send should not be called after requestObserver.onCompleted() is called.
requestObserver.onCompleted() is called when the stream is closed.

I am investigating stalled GetWork streams with logs like Output channel stalled for 31s, outbound thread. I think sending messages on a closed stream has something to do with it. In test runs with this change, I'm yet to see a stalled GetWorkStream.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

arunpandianp · 2022-03-23T03:48:11Z

R: @scwhittle

scwhittle · 2022-03-23T08:18:26Z

...orker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/GrpcWindmillServer.java

@@ -936,7 +936,11 @@ protected void onResponse(StreamingGetWorkResponseChunk chunk) {
              .execute(
                  () -> {
                    try {
-                      send(extension);
+                      synchronized (this) {
+                        if (!clientClosed.get()) {


Should send itself be changed to check this?

Most sends are wrapped with
catch (IllegalStateException)
but then ignore the exception or just log.

Was this causing issues other than unnecessary IllegalStateExceptions?

Was this causing issues other than unnecessary IllegalStateExceptions?

I am investigating stalled GetWork streams with logs like Output channel stalled for 31s, outbound thread. I think sending messages on a closed stream has something to do with it. In test runs with this change, I'm yet to see a stalled GetWorkStream.

Should send itself be changed to check this?

There are a few sends without the catch(IllegalStateException), they need to be looked at individually. Can we get this in first?

To keep contract the same for other call-sites you could throw IllegalStateException yourself if the client is closed instead of relying on grpc to do so (since that adds delay it sounds like).

LGTM though if you want to put a comment/TODO for that to try to get this in the next release. You'll have to find someone else to merge though.

+1 to throwing IllegalStateException if the stream is closed. Otherwise we risk having regressions by changing the behavior. It's safer to throw the exception for the purposes of your change.

Moved the closed check inside send.

…treams requestObserver.onNext() which is called by send should not be called after requestObserver.onCompleted() is called. requestObserver.onCompleted() is called when the stream is closed. I am investigating stalled windmill streams with logs like `Output channel stalled for 31s, outbound thread`. I think sending messages on a closed stream has something to do with it. In test runs with this change, I'm yet to see a stalled GetWorkStream.

arunpandianp · 2022-03-24T10:13:45Z

R: @TheNeuralBit

@TheNeuralBit could you help review this?

scwhittle · 2022-03-24T11:32:46Z

Run Java PreCommit

scwhittle · 2022-03-24T11:32:54Z

LGTM, thanks!

TheNeuralBit

Note you just missed the 2.38.0 release cut, what is the severity of the issue this is fixing?

TheNeuralBit · 2022-03-24T15:43:01Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

+              try {
+                refreshActiveWork();
+              } catch (RuntimeException e) {
+                LOG.warn("Failed to refresh active work: ", e);


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?

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

Got it, thank you!

…ient closed check This is a follow up to apache#17162. An AbstractWindmillStream can have more than one grpc stream during its lifetime, new streams can be created after client closed for sending pending requests. So it is not correct to check `if(clientClosed)` in `send()`, this PR adds a new grpc stream level boolean to do the closed check in `send()`.

…ient closed check (#17191) * [BEAM-14157] GrpcWindmillServer: Use stream specific boolean to do client closed check This is a follow up to #17162. An AbstractWindmillStream can have more than one grpc stream during its lifetime, new streams can be created after client closed for sending pending requests. So it is not correct to check `if(clientClosed)` in `send()`, this PR adds a new grpc stream level boolean to do the closed check in `send()`. * [BEAM-14157] Add unit test testing CommitWorkStream retries around stream closing * [BEAM-14157] review comments * [BEAM-14157] review comments * [BEAM-14157] review comments * [BEAM-14157] fix test * [BEAM-14157] fix test Co-authored-by: Arun Pandian <pandiana@google.com>

…ient closed check (apache#17191) * [BEAM-14157] GrpcWindmillServer: Use stream specific boolean to do client closed check This is a follow up to apache#17162. An AbstractWindmillStream can have more than one grpc stream during its lifetime, new streams can be created after client closed for sending pending requests. So it is not correct to check `if(clientClosed)` in `send()`, this PR adds a new grpc stream level boolean to do the closed check in `send()`. * [BEAM-14157] Add unit test testing CommitWorkStream retries around stream closing * [BEAM-14157] review comments * [BEAM-14157] review comments * [BEAM-14157] review comments * [BEAM-14157] fix test * [BEAM-14157] fix test Co-authored-by: Arun Pandian <pandiana@google.com>

* added initial commit * removed modified file * removed params that dont exist * added clock, removed generics that were causing pickle error, fixed metrics name * fixed class names removed class that goes in apis * added base test file * Added unit tests * reordered imports * replied to comments * apis to api * added license * added mock clock test for metrics, realized our metric wouldn't work right with a generator * Minor changes from Andys comments. Push metric namespace decision to modleLoader class * Update sdks/python/apache_beam/ml/inference/base.py typo fix valentyn's suggestion Co-authored-by: tvalentyn <tvalentyn@users.noreply.github.com> * updated with changes from valentyns comments * merged from tfx version * added comment * linted * changed import order for jenkins linter * added a bug to track a todo * fixed for Roberts comments * make clock and metrics collector private * make shared second parameter * mark RunInferenceDoFn private * moved initialization of shared.Shared into constructor * added todo * Update sdks/python/apache_beam/ml/inference/base.py Co-authored-by: Brian Hulette <hulettbh@gmail.com> * Update sdks/python/apache_beam/ml/inference/base.py Co-authored-by: Brian Hulette <hulettbh@gmail.com> * Update sdks/python/apache_beam/ml/inference/base.py Co-authored-by: Brian Hulette <hulettbh@gmail.com> * updated to correct variable names * udpated variable names * added typevar * remove unbatch * added note that users should expect changes * Update python container version * Add --dataflowServiceOptions=enable_prime to useUnifiedWorker conditions (#17213) * Add self-descriptive message for expected errors. Ideally we would not log these in the first place, but this is an easy hack. * [BEAM-10529] nullable xlang coder (#16923) * [BEAM-10529] add java and generic components of nullable xlang tests * [BEAM-10529] fix test case * [BEAM-10529] add coders and typehints to support nullable xlang coders * [BEAM-10529] update external builder to support nullable coder * [BEAM-10529] clean up coders.py * [BEAM-10529] add coder translation test * [BEAM-10529] add additional check to typecoder to not accidentally misidentify coders as nullable * [BEAM-10529] add test to retrieve nullable coder from typehint * [BEAM-10529] run spotless * [BEAM-10529] add go nullable coder * [BEAM-10529] cleanup extra println * [BEAM-10529] improve comments, clean up python * [BEAM-10529] remove changes to kafkaIO to simplify pr * [BEAM-10529] add coders to go exec, add asf license text * [BEAM-10529] clean up error handlign * [BEAM-10529] update go fromyaml to handle nullable cases * [BEAM-10529] add unit test, register nullable coder in dataflow.go * [BEAM-10529] remove mistaken commit * [BEAM-10529] add argument check to CoderTranslators * [BEAM-10529] Address python comments & cleanup * [BEAM-10529] address go comments * [BEAM-10529] remove extra check that was added in error * [BEAM-10529] fix typo * [BEAM-10529] re-order check for nonetype to prevent attribute errors * [BEAM-10529] change isinstance to == * Fix go fmt break in core/typex/special.go (#17266) * [BEAM-8970] Add docs to run wordcount example on portable Spark Runner * [BEAM-8970] Add period to end of sentence * [BEAM-5436] Add doc page on Go cross compilation. (#17256) * Pr-bot Don't count all reviews as approvals (#17269) * Fix postcommits (#17263) * [BEAM-14241] Address staticcheck warnings in boot.go (#17264) * [BEAM-14157] GrpcWindmillServer: Use stream specific boolean to do client closed check (#17191) * [BEAM-14157] GrpcWindmillServer: Use stream specific boolean to do client closed check This is a follow up to #17162. An AbstractWindmillStream can have more than one grpc stream during its lifetime, new streams can be created after client closed for sending pending requests. So it is not correct to check `if(clientClosed)` in `send()`, this PR adds a new grpc stream level boolean to do the closed check in `send()`. * [BEAM-14157] Add unit test testing CommitWorkStream retries around stream closing * [BEAM-14157] review comments * [BEAM-14157] review comments * [BEAM-14157] review comments * [BEAM-14157] fix test * [BEAM-14157] fix test Co-authored-by: Arun Pandian <pandiana@google.com> * [BEAM-10582] Allow (and test) pyarrow 7 (#17229) * [BEAM-13519] Solve race issues when the server responds with an error before the GrpcStateClient finishes being constructed. (#17240) * [BEAM-13519] Solve race issues when the server responds with an error before the GrpcStateClient finishes. The issue was that the InboundObserver can be invoked before outboundObserverFactory#outboundObserverFor returns meaning that the server is waiting for a response for cache.remove but cache.computeIfAbsent is being invoked at the same time. Another issue was that the outstandingRequests map could be updated with another request within GrpcStateClient during closeAndCleanup meaning that the CompleteableFuture would never be completed exceptionally. Passes 1000 times locally now without getting stuck or failing. * [BEAM-14256] update SpEL dependency to 5.3.18.RELEASE * [BEAM-14256] remove .RELEASE * [BEAM-13015] Disable retries for fnapi grpc channels which otherwise defaults on. (#17243) * [BEAM-13015] Disable retries for grpc channels which otherwise default to true. Since the channel is to the local runner process, retries are not expected to help. This simplifies the grpc stream stack to not involve a RetryStream object. * fixup comment * Update sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/channel/ManagedChannelFactory.java * Update sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/channel/ManagedChannelFactory.java Co-authored-by: Lukasz Cwik <lcwik@google.com> * [BEAM-9649] Add region option to Mongo Dataflow test. * Fix dependency issue causing failures * [BEAM-13952] Sickbay testAfterProcessingTimeContinuationTriggerUsingState (#17214) * BEAM-14235 parquetio module does not parse PEP-440 compliant Pyarrow version (#17275) * Update parquetio.py * Update CHANGES.md * Fix import order * [BEAM-14250] Fix coder registration for types defined in __main__. Until all runners are portable and we can get rid of all round trips between Pipeline and proto representatons, register types in __main__ according to their string representations as pickling does not preserve identity. * Allow get_coder(None). Co-authored-by: Andy Ye <andyye333@gmail.com> * [Website] Contribution guide page indent bug fix (#17287) * Fix markdown indent issue in Development Setup section * update query * [BEAM-10976] Document go sdk bundle finalization (#17048) * [BEAM-13829] Expose status API from Go SDK Harness (#16957) * Avoid pr-bot state desync (#17299) * [BEAM-14259] Clean up staticcheck warnings in the exec package (#17285) * Minor: Prefer registered schema in SQL docs (#17298) * Prefer registered schema in SQL docs * address review comments * [Playground] add meta tags (#17207) * playground add meta tags * playground fix meta tags * fixes golint and deprecated issues in recent Go SDK import (#17304) * [BEAM-14262] Update plugins for Dockerized Jenkins. I copied the list from the cwiki and removed all of the ones that failed to install. https://cwiki.apache.org/confluence/display/INFRA/ci-beam.apache.org * Add ansicolor and ws-cleanup plugins. Without them, the seed job prints warnings: Warning: (CommonJobProperties.groovy, line 107) plugin 'ansicolor' needs to be installed Warning: (CommonJobProperties.groovy, line 113) plugin 'ws-cleanup' needs to be installed * [BEAM-14266] Replace deprecated ptypes package uses (#17302) * [BEAM-11936] Fix rawtypes warnings in SnowflakeIO (#17257) * [BEAM-10556] Fix rawtypes warnings in SnowflakeIO * fixup! [BEAM-10556] Fix rawtypes warnings in SnowflakeIO * Merge pull request #17262: [BEAM-14244] Use the supplied output timestamp for processing time timers rather than the input watermark * removed unused typing * added list typing * linted Co-authored-by: tvalentyn <tvalentyn@users.noreply.github.com> Co-authored-by: Brian Hulette <hulettbh@gmail.com> Co-authored-by: kileys <kileysok@google.com> Co-authored-by: Yichi Zhang <zyichi@google.com> Co-authored-by: Kyle Weaver <kcweaver@google.com> Co-authored-by: johnjcasey <95318300+johnjcasey@users.noreply.github.com> Co-authored-by: Jack McCluskey <34928439+jrmccluskey@users.noreply.github.com> Co-authored-by: Benjamin Gonzalez <benjamin.gonzalez@wizeline.com> Co-authored-by: Robert Burke <lostluck@users.noreply.github.com> Co-authored-by: Danny McCormick <dannymccormick@google.com> Co-authored-by: Arun Pandian <arunpandianp@gmail.com> Co-authored-by: Arun Pandian <pandiana@google.com> Co-authored-by: Brian Hulette <bhulette@google.com> Co-authored-by: Lukasz Cwik <lcwik@google.com> Co-authored-by: johnjcasey <johnjcasey@google.com> Co-authored-by: scwhittle <scwhittle@users.noreply.github.com> Co-authored-by: Arwin Tio <arwin.tio@adroll.com> Co-authored-by: Robert Bradshaw <robertwb@gmail.com> Co-authored-by: Andy Ye <andyye333@gmail.com> Co-authored-by: Yi Hu <yathu@google.com> Co-authored-by: Michael Li <bingyeli@google.com> Co-authored-by: Ritesh Ghorse <riteshghorse@gmail.com> Co-authored-by: Aydar Farrakhov <stranniknm@gmail.com> Co-authored-by: Kamil Breguła <kamil.bregula@snowflake.com> Co-authored-by: Steven Niemitz <steveniemitz@gmail.com>

github-actions bot added dataflow runners labels Mar 23, 2022

scwhittle reviewed Mar 23, 2022

View reviewed changes

arunpandianp force-pushed the BEAM-14157 branch 2 times, most recently from a0098f3 to 3c73617 Compare March 23, 2022 21:04

arunpandianp force-pushed the BEAM-14157 branch from 3c73617 to 4411f01 Compare March 23, 2022 21:06

arunpandianp changed the title ~~[BEAM-14157] Don't request work on a closed windmill GetWorkStream~~ [BEAM-14157] Don't call requestObserver.onNext on closed windmill streams Mar 23, 2022

TheNeuralBit reviewed Mar 24, 2022

View reviewed changes

TheNeuralBit approved these changes Mar 24, 2022

View reviewed changes

TheNeuralBit merged commit 993061a into apache:master Mar 24, 2022

arunpandianp mentioned this pull request Mar 26, 2022

[BEAM-14157] GrpcWindmillServer: Use stream specific boolean to do client closed check #17191

Merged

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

arunpandianp commented Mar 23, 2022 •

edited

Loading

arunpandianp commented Mar 23, 2022

scwhittle Mar 23, 2022

arunpandianp Mar 23, 2022

scwhittle Mar 23, 2022

ayberk Mar 23, 2022 •

edited

Loading

arunpandianp Mar 24, 2022

arunpandianp commented Mar 24, 2022

scwhittle commented Mar 24, 2022

scwhittle commented Mar 24, 2022

TheNeuralBit left a comment

TheNeuralBit Mar 24, 2022

arunpandianp Mar 24, 2022

TheNeuralBit Mar 24, 2022

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

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

Conversation

arunpandianp commented Mar 23, 2022 • edited Loading

GitHub Actions Tests Status (on master branch)

arunpandianp commented Mar 23, 2022

scwhittle Mar 23, 2022

Choose a reason for hiding this comment

arunpandianp Mar 23, 2022

Choose a reason for hiding this comment

scwhittle Mar 23, 2022

Choose a reason for hiding this comment

ayberk Mar 23, 2022 • edited Loading

Choose a reason for hiding this comment

arunpandianp Mar 24, 2022

Choose a reason for hiding this comment

arunpandianp commented Mar 24, 2022

scwhittle commented Mar 24, 2022

scwhittle commented Mar 24, 2022

TheNeuralBit left a comment

Choose a reason for hiding this comment

TheNeuralBit Mar 24, 2022

Choose a reason for hiding this comment

arunpandianp Mar 24, 2022

Choose a reason for hiding this comment

TheNeuralBit Mar 24, 2022

Choose a reason for hiding this comment

arunpandianp commented Mar 23, 2022 •

edited

Loading

ayberk Mar 23, 2022 •

edited

Loading