github-actions[bot] commented on code in PR #40606:
URL: https://github.com/apache/beam/pull/40606#discussion_r4300114262
##########
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:
**🟠High**: When elements arrive before `registerConsumer` is called,
`receivers` already contains a `ReceiverHolder` (`existing`) with buffered
elements. If `complete(existing, receiver)` throws an exception because
`receiver.accept(p.elements)` fails while draining `pending` (as in
`testPermitsReleasedWhenReceiverThrowsOnBufferedElements`),
`existing.complete(receiver)` has already set `this.receiver = receiver` (line
167), and the thrown exception aborts `receivers.compute(...)`.
Per `ConcurrentHashMap.compute` semantics, throwing from the remapping
function leaves the existing entry in `receivers`. As a result,
`hasConsumer(instructionId)` remains `true`, `instructionId` is not poisoned,
`receiver.close()` is not called (unlike when `receiver.accept` throws during
`onNext` in `forwardToConsumerForInstructionId`), and `receivers.size()` stays
permanently inflated by the leaked entry (which also inflates `maxWindow` in
`onStalledPoll()`).
Consider catching failures from `holder.complete(receiver)` and calling
`poisonInstructionId(instructionId)` outside `receivers.compute` before
rethrowing the exception.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserver.java:
##########
@@ -177,19 +200,28 @@ public void awaitCompletion() throws Exception {
// collection.
Review Comment:
**🟡 Medium**: Two issues here:
1. The `{ ... }` block on lines 201–207 exists specifically so no local
variable in the outer `while` scope retains a reference to `BeamFnApi.Elements`
while `multiplexElements(dataIterator, timersIterator)` runs (see the comment
on lines 195–198 and `createDiscardingIterator`). Declaring
`ElementWithCallback wrapper` outside `{ ... }` and calling
`wrapper.callback.run()` on line 215 keeps `wrapper` (and its `final
BeamFnApi.Elements elements` field) strongly reachable for the entire duration
of `multiplexElements`. Since `inProgressCallback = wrapper.callback` is
already captured inside `{ ... }`, `wrapper` can remain scoped inside `{ ... }`
and lines 214–215 can invoke `inProgressCallback` after clearing it.
2. `if (inProgressCallback != null) { inProgressCallback.run(); }` is only
executed in `catch (Exception e)` (lines 222–224). If `multiplexElements`
throws an `Error` (e.g., `AssertionError`), `finally` cancels the queue via
`close()` but leaks the `inProgressCallback` permit. Moving the
`inProgressCallback` cleanup into `finally` (after `close()`) covers all
`Throwable`s.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/fn/stream/DirectStreamObserver.java:
##########
@@ -55,60 +56,80 @@ public final class DirectStreamObserver<T> implements
StreamObserver<T> {
private final Object lock = new Object();
private int numMessages;
+ private final @Nullable StalledPollHandler stalledPollHandler;
+
public DirectStreamObserver(Phaser phaser, CallStreamObserver<T>
outboundObserver) {
- this(phaser, outboundObserver, DEFAULT_MAX_MESSAGES_BEFORE_CHECK);
+ this(phaser, outboundObserver, null);
+ }
+
+ public DirectStreamObserver(
+ Phaser phaser,
+ CallStreamObserver<T> outboundObserver,
+ @Nullable StalledPollHandler stalledPollHandler) {
+
+ this(phaser, outboundObserver, DEFAULT_MAX_MESSAGES_BEFORE_CHECK,
stalledPollHandler);
}
DirectStreamObserver(
Phaser phaser, CallStreamObserver<T> outboundObserver, int
maxMessagesBeforeCheck) {
+ this(phaser, outboundObserver, maxMessagesBeforeCheck, null);
+ }
+
+ DirectStreamObserver(
+ Phaser phaser,
+ CallStreamObserver<T> outboundObserver,
+ int maxMessagesBeforeCheck,
+ @Nullable StalledPollHandler stalledPollHandler) {
this.phaser = phaser;
this.outboundObserver = outboundObserver;
this.maxMessagesBeforeCheck = maxMessagesBeforeCheck;
+ this.stalledPollHandler = stalledPollHandler;
}
@Override
public void onNext(T value) {
synchronized (lock) {
if (numMessages >= maxMessagesBeforeCheck) {
numMessages = 0;
- int waitSeconds = 1;
- int totalSecondsWaited = 0;
+ int waitMs = 50;
+ int totalMsWaited = 0;
int phase = phaser.getPhase();
// Record the initial phase in case we are in the inbound gRPC thread
where the phase won't
// advance.
int initialPhase = phase;
// A negative phase indicates that the phaser is terminated.
while (phase >= 0 && !outboundObserver.isReady()) {
try {
- phase = phaser.awaitAdvanceInterruptibly(phase, waitSeconds,
TimeUnit.SECONDS);
+ phase = phaser.awaitAdvanceInterruptibly(phase, waitMs,
TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
- totalSecondsWaited += waitSeconds;
- // Double the backoff for re-evaluating the isReady bit up to a
maximum of once per
- // minute. This bounds the waiting if the onReady callback is not
called as expected.
- waitSeconds = Math.min(waitSeconds * 2, 60);
+ totalMsWaited += waitMs;
+ if (stalledPollHandler != null && totalMsWaited >= 500 &&
totalMsWaited % 500 == 0) {
+ stalledPollHandler.onStalledPoll();
+ }
+ waitMs = 50;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(e);
}
}
- if (totalSecondsWaited > 0) {
+ if (totalMsWaited > 0) {
Review Comment:
**🟡 Medium**: Because `waitMs` was reduced from `1s` to `50ms` and
`DEFAULT_MAX_MESSAGES_BEFORE_CHECK` from `100` to `0`, `totalMsWaited > 0` now
triggers after a single 50ms poll. For any wait between 50ms and 950ms,
`totalMsWaited / 1000` evaluates to `0` (`"Output channel stalled for 0s"`).
Furthermore, under normal backpressure, `outboundObserver.isReady()` can become
`true` during a 50ms poll slightly before the gRPC executor dispatches the
`onReady` callback (`phaser.arrive()`), exiting the loop with `totalMsWaited ==
50` and `initialPhase == phase` and logging a false-positive `LOG.info`
warning. Requiring `totalMsWaited >= 1000` preserves the previous 1-second
minimum threshold for stall logging and avoids logging `0s`.
```suggestion
if (totalMsWaited >= 1000) {
```
--
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]
