Support poisioning instruction ids to prevent the FnApi data stream from blocking on failed instructions #32857

scwhittle · 2024-10-18T10:07:53Z

Previously it could occur that an instruction id was observed on the control stream but due to exception it would never register a handler and the data stream would be forever blocked trying to multiplex elements to it's queue.

Fixes #32714

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

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
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 or the workflows README to see a list of phrases to trigger workflows.

github-actions · 2024-10-18T11:34:41Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

scwhittle · 2024-10-21T11:12:05Z

Test failures appear to be unrelated flakiness

scwhittle · 2024-10-21T11:12:08Z

assign set of reviewers

scwhittle · 2024-10-21T11:12:51Z

Run Java PreCommit

github-actions · 2024-10-21T11:14:17Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @robertwb for label java.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

scwhittle · 2024-10-23T08:25:23Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/data/BeamFnDataGrpcClient.java

+  @Override
+  public void poisonInstructionId(String instructionId) {
+    LOG.debug("Poisoning instruction {}", instructionId);
+    for (BeamFnDataGrpcMultiplexer client : multiplexerCache.values()) {


one thing this doesn't yet handle is if the instruction that is being poisoned is for an endpoint that has not yet been observed. This is only possible at beginning of the processing though.

I was working on refactoring creation of the bundle processor to determine the api endpoints before possible exception was thrown but it was getting large and would prefer to do it as a follow up because as-is this helps cases where the api endpoint was created before.

+1 to getting this in first.

An alternative is for all these multiplexers to share the same poisoned id cache.

robertwb

Thanks for hunting this down and fixing it.

robertwb · 2024-10-24T00:24:08Z

sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java

    this.receivers = new ConcurrentHashMap<>();
-    this.erroredInstructionIds = new ConcurrentHashMap<>();
+    this.poisonedInstructionIds =
+        CacheBuilder.newBuilder().expireAfterWrite(Duration.ofMinutes(20)).build();


Should 20 minutes be pulled out to a top-level constant?

robertwb · 2024-10-24T00:30:41Z

sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java

+        instructionId,
+        (unused) -> {
+          if (poisonedInstructionIds.getIfPresent(instructionId) != null) {
+            return null;


Why do we have to special case this instead of simply always returning a completable future (even if it won't be used) as before? Or is it for better error messaging?

This null in the computeIfAbsent is to avoid memory leak for long-running pipelines processing lots of instructions ids. If we're poisoned, it seems possible we won't unregister and thus if we add the id here it will stay in the map forever.

I removed this helper since I think it is clearer if the two cases do their own handling.

robertwb · 2024-10-24T00:34:57Z

sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java

+    CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture =
+        receivers.remove(instructionId);
+    if (receiverFuture != null && !receiverFuture.isDone()) {
+      throw new IllegalStateException("Unregistering consumer which was not registered.");


This error message doesn't seem to match the clauses of the if statement.

Added explanation comment.

robertwb · 2024-10-24T00:43:29Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/data/BeamFnDataGrpcClient.java

+  @Override
+  public void poisonInstructionId(String instructionId) {
+    LOG.debug("Poisoning instruction {}", instructionId);
+    for (BeamFnDataGrpcMultiplexer client : multiplexerCache.values()) {


+1 to getting this in first.

An alternative is for all these multiplexers to share the same poisoned id cache.

codecov · 2024-10-25T11:56:55Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 58.95%. Comparing base (1e27978) to head (abc5492).
Report is 121 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #32857      +/-   ##
============================================
+ Coverage     57.41%   58.95%   +1.53%     
- Complexity     1475     3112    +1637     
============================================
  Files           968     1131     +163     
  Lines        154224   174787   +20563     
  Branches       1076     3357    +2281     
============================================
+ Hits          88546   103038   +14492     
- Misses        63477    68398    +4921     
- Partials       2201     3351    +1150

Flag	Coverage Δ
java	`69.94% <ø> (+1.35%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

robertwb

LGTM once tests are passing.

scwhittle · 2024-10-31T14:45:42Z

Run Java PreCommit

scwhittle · 2024-10-31T14:45:50Z

Run Java_IOs_Direct PreCommit

scwhittle · 2024-10-31T14:46:19Z

Run Java_Pulsar_IO_Direct PreCommit

scwhittle · 2024-11-11T16:15:38Z

Run Java_IOs_Direct PreCommit

scwhittle · 2024-11-11T16:15:55Z

Run Java_Hadoop_IO_Direct PreCommit

scwhittle · 2024-11-11T16:16:02Z

Run Java PreCommit

scwhittle · 2024-11-11T16:17:28Z

Run Java_Hadoop_IO_Direct PreCommit

…rom blocking on failed instructions

…iterable: - remove reference to completed encoded input page from decoder once we have read it. - re-read from cache after loading the next page to give eviction a chance to remove blocks

scwhittle · 2024-11-11T21:02:11Z

(removed, commented on wrong issue)

scwhittle · 2024-11-11T21:04:35Z

Run Java PreCommit

scwhittle · 2024-11-11T21:04:45Z

Run Java_IOs_Direct PreCommit

scwhittle · 2024-11-11T21:04:58Z

Run Java_Pulsar_IO_Direct PreCommit

scwhittle · 2024-11-12T09:46:15Z

Run Java PreCommit

scwhittle · 2024-11-12T09:46:43Z

Run Java_IOs_Direct PreCommit

scwhittle · 2024-11-12T10:36:24Z

precommit failures were unrelated flakes in :runners:google-cloud-dataflow-java:worker:test

github-actions bot added the java label Oct 18, 2024

scwhittle marked this pull request as ready for review October 18, 2024 10:08

scwhittle force-pushed the timeout_and_poison branch 3 times, most recently from 974a9b4 to fa7cde6 Compare October 21, 2024 10:21

github-actions bot added the Next Action: Reviewers label Oct 21, 2024

scwhittle commented Oct 23, 2024

View reviewed changes

robertwb reviewed Oct 24, 2024

View reviewed changes

robertwb approved these changes Oct 29, 2024

View reviewed changes

scwhittle closed this Oct 30, 2024

scwhittle reopened this Oct 30, 2024

scwhittle closed this Nov 4, 2024

scwhittle reopened this Nov 4, 2024

scwhittle closed this Nov 5, 2024

scwhittle reopened this Nov 5, 2024

scwhittle force-pushed the timeout_and_poison branch from 82fb606 to 49d08b4 Compare November 6, 2024 08:52

scwhittle closed this Nov 11, 2024

scwhittle reopened this Nov 11, 2024

scwhittle force-pushed the timeout_and_poison branch from 49d08b4 to 7386855 Compare November 11, 2024 10:07

scwhittle added 3 commits November 11, 2024 19:15

Support poisioning instruction ids to prevent the FnApi data stream f…

50ea4ce

…rom blocking on failed instructions

address pr comments

27908d5

Changes to reduce memory pinned while iterating through state backed …

d410d20

…iterable: - remove reference to completed encoded input page from decoder once we have read it. - re-read from cache after loading the next page to give eviction a chance to remove blocks

scwhittle force-pushed the timeout_and_poison branch from 7386855 to d410d20 Compare November 11, 2024 18:15

scwhittle merged commit 785ec07 into apache:master Nov 12, 2024

scwhittle deleted the timeout_and_poison branch November 12, 2024 10:36

Support poisioning instruction ids to prevent the FnApi data stream from blocking on failed instructions #32857

Support poisioning instruction ids to prevent the FnApi data stream from blocking on failed instructions #32857

Uh oh!

Conversation

scwhittle commented Oct 18, 2024

GitHub Actions Tests Status (on master branch)

Uh oh!

github-actions bot commented Oct 18, 2024

Uh oh!

scwhittle commented Oct 21, 2024

Uh oh!

scwhittle commented Oct 21, 2024

Uh oh!

scwhittle commented Oct 21, 2024

Uh oh!

github-actions bot commented Oct 21, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

robertwb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov bot commented Oct 25, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

robertwb left a comment

Choose a reason for hiding this comment

Uh oh!

scwhittle commented Oct 31, 2024

Uh oh!

scwhittle commented Oct 31, 2024

Uh oh!

scwhittle commented Oct 31, 2024

Uh oh!

scwhittle commented Nov 11, 2024

Uh oh!

scwhittle commented Nov 11, 2024

Uh oh!

scwhittle commented Nov 11, 2024

Uh oh!

scwhittle commented Nov 11, 2024

Uh oh!

scwhittle commented Nov 11, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

scwhittle commented Nov 11, 2024

Uh oh!

scwhittle commented Nov 11, 2024

Uh oh!

scwhittle commented Nov 11, 2024

Uh oh!

scwhittle commented Nov 12, 2024

Uh oh!

scwhittle commented Nov 12, 2024

Uh oh!

scwhittle commented Nov 12, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

codecov bot commented Oct 25, 2024 •

edited

Loading

scwhittle commented Nov 11, 2024 •

edited

Loading