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

…rom blocking on failed instructions

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

@@ -64,7 +75,8 @@ public BeamFnDataGrpcMultiplexer(
          baseOutboundObserverFactory) {
    this.apiServiceDescriptor = apiServiceDescriptor;
    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.

…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-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

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 2 times, most recently from caf0eaa to 974a9b4 Compare October 21, 2024 09:21

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

fa7cde6

…rom blocking on failed instructions

scwhittle force-pushed the timeout_and_poison branch 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

address pr comments

abc5492

robertwb approved these changes Oct 29, 2024

View reviewed changes

scwhittle closed this Oct 30, 2024

scwhittle reopened this Oct 30, 2024

Changes to reduce memory pinned while iterating through state backed …

82fb606

…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 closed this Nov 4, 2024

scwhittle reopened this Nov 4, 2024

scwhittle closed this Nov 5, 2024

scwhittle reopened this Nov 5, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

scwhittle commented Oct 18, 2024

github-actions bot commented Oct 18, 2024

scwhittle commented Oct 21, 2024

scwhittle commented Oct 21, 2024

scwhittle commented Oct 21, 2024

github-actions bot commented Oct 21, 2024

scwhittle Oct 23, 2024

robertwb Oct 24, 2024

robertwb left a comment

robertwb Oct 24, 2024

scwhittle Oct 25, 2024

robertwb Oct 24, 2024

scwhittle Oct 25, 2024

robertwb Oct 24, 2024

scwhittle Oct 25, 2024

robertwb Oct 24, 2024

codecov bot commented Oct 25, 2024 •

edited

Loading

robertwb left a comment

scwhittle commented Oct 31, 2024

scwhittle commented Oct 31, 2024

scwhittle commented Oct 31, 2024

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

Are you sure you want to change the base?

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

Conversation

scwhittle commented Oct 18, 2024

GitHub Actions Tests Status (on master branch)

github-actions bot commented Oct 18, 2024

scwhittle commented Oct 21, 2024

scwhittle commented Oct 21, 2024

scwhittle commented Oct 21, 2024

github-actions bot commented Oct 21, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robertwb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Oct 25, 2024 • edited Loading

Codecov Report

robertwb left a comment

Choose a reason for hiding this comment

scwhittle commented Oct 31, 2024

scwhittle commented Oct 31, 2024

scwhittle commented Oct 31, 2024

codecov bot commented Oct 25, 2024 •

edited

Loading