prodriguezdefino commented on code in PR #40606:
URL: https://github.com/apache/beam/pull/40606#discussion_r4300440776
##########
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:
Fixed in 6f9d444 with `totalMsWaited >= 1000`, as suggested.
The stall hook still runs every 500 ms.
- `testShortStallIsNotLogged` checks that a 300 ms stall logs nothing,
- and `testStallOfASecondIsLogged` checks that a 1.3 s stall still logs
"stalled for 1s".
--
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]
