prodriguezdefino commented on code in PR #40606:
URL: https://github.com/apache/beam/pull/40606#discussion_r4300434291
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java:
##########
@@ -118,26 +280,45 @@ public void registerConsumer(
instructionId,
(unused, existing) -> {
if (existing != null) {
- if (!existing.complete(receiver)) {
+ if (existing.isPoisoned()) {
+ throw new IllegalArgumentException("Instruction id was
poisoned");
+ }
+ if (!complete(existing, receiver)) {
Review Comment:
Fixed in 6f9d444.
When the receiver throws on a buffered element, `registerConsumer` poisons
the instruction after `compute()` returns, which removes the holder and closes
the receiver, as a failure on the inbound gRPC thread does. A duplicate or
poisoned registration still only throws, without poisoning.
- `testPermitsReleasedWhenReceiverThrowsOnBufferedElements` now also checks
that the entry is gone, that the receiver was closed, and that a later
registration fails as poisoned.
- `testDuplicateRegistrationDoesNotPoison` checks that a rejected duplicate
leaves the first registration working.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
