You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "m-trieu (via GitHub)" <gi...@apache.org> on 2025/03/13 19:22:19 UTC
[PR] do not block other threads from health checking if a stream is blocked [beam]
m-trieu opened a new pull request, #34283:
URL: https://github.com/apache/beam/pull/34283
Currently a single blocked stream can prevent health checks from multiple streams. Use the stream's internal executor to hand do the health check RPC in a way that does not block the health check scheduling thread.
R: @acrites @scwhittle
------------------------
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](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #34283:
URL: https://github.com/apache/beam/pull/34283
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2015054873
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
+ sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.unblockSend();
+
+ // Make sure the future completes.
+ sendFuture.get();
+
+ assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
+ testStream.shutdown();
+ }
+
+ @Test
+ public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
+ throws ExecutionException, InterruptedException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Future<?> sendFuture =
+ sendExecutor.submit(
+ () -> {
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // Set a really long reporting threshold.
+ Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
+ // Should not send health checks since we just sent the above message.
+ testStream.maybeSendHealthCheck(reportingThreshold);
+ testStream.maybeSendHealthCheck(reportingThreshold);
+ });
+
+ callStreamObserver.unblockSend();
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2025843072
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +80,65 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2015043982
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2023491485
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -197,18 +202,37 @@ protected void shutdownInternal() {}
}
private static class TestCallStreamObserver extends CallStreamObserver<Integer> {
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractWindmillStreamTest.class);
private final CountDownLatch sendBlocker = new CountDownLatch(1);
+ private final boolean waitForSend;
+
+ private TestCallStreamObserver(boolean waitForSend) {
+ this.waitForSend = waitForSend;
+ }
+
private void unblockSend() {
sendBlocker.countDown();
}
- @Override
- public void onNext(Integer integer) {
+ private void waitForSend() {
try {
- sendBlocker.await();
+ int waitedMillis = 0;
+ while (!sendBlocker.await(100, TimeUnit.MILLISECONDS)) {
+ waitedMillis += 100;
+ LOG.info("Waiting from send for {}ms", waitedMillis);
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2015055682
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
+ sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.unblockSend();
+
+ // Make sure the future completes.
+ sendFuture.get();
+
+ assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
+ testStream.shutdown();
+ }
+
+ @Test
+ public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
+ throws ExecutionException, InterruptedException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Future<?> sendFuture =
+ sendExecutor.submit(
+ () -> {
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // Set a really long reporting threshold.
+ Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
+ // Should not send health checks since we just sent the above message.
+ testStream.maybeSendHealthCheck(reportingThreshold);
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2012822232
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java:
##########
@@ -236,13 +239,34 @@ protected final void executeSafely(Runnable runnable) {
}
}
- public final synchronized void maybeSendHealthCheck(Instant lastSendThreshold) {
- if (!clientClosed && debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()) {
- try {
- sendHealthCheck();
- } catch (Exception e) {
- logger.debug("Received exception sending health check.", e);
- }
+ /**
+ * Schedule an application level keep-alive health check to be sent on the stream.
+ *
+ * @implNote This is sent asynchronously via an executor to minimize blocking. Messages are sent
+ * serially. If we recently sent a message before we attempt to schedule the health check, the
+ * stream has been restarted/closed, there is an active health check that hasn't completed due
+ * to flow control/pushback or there was a more recent send by the time we enter the
+ * synchronized block, we skip the attempt to send scheduled the health check.
+ */
+ public final void maybeSendHealthCheck(Instant lastSendThreshold) {
+ if (debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis() && !isHealthCheckActive) {
Review Comment:
done, opted to use an AtomicBoolean
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2015050399
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
+ sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.unblockSend();
+
+ // Make sure the future completes.
+ sendFuture.get();
+
+ assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2026148522
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +80,65 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Review Comment:
fixed, made it reflect the real behavior by using `isReady`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2015065916
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
+ sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.unblockSend();
+
+ // Make sure the future completes.
+ sendFuture.get();
+
+ assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
+ testStream.shutdown();
+ }
+
+ @Test
+ public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
+ throws ExecutionException, InterruptedException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Future<?> sendFuture =
+ sendExecutor.submit(
+ () -> {
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // Set a really long reporting threshold.
+ Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
+ // Should not send health checks since we just sent the above message.
+ testStream.maybeSendHealthCheck(reportingThreshold);
+ testStream.maybeSendHealthCheck(reportingThreshold);
+ });
+
+ callStreamObserver.unblockSend();
Review Comment:
done
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
+ sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.unblockSend();
+
+ // Make sure the future completes.
+ sendFuture.get();
+
+ assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
+ testStream.shutdown();
+ }
+
+ @Test
+ public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
+ throws ExecutionException, InterruptedException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Future<?> sendFuture =
+ sendExecutor.submit(
+ () -> {
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // Set a really long reporting threshold.
+ Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
+ // Should not send health checks since we just sent the above message.
+ testStream.maybeSendHealthCheck(reportingThreshold);
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2023477433
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java:
##########
@@ -244,27 +244,28 @@ protected final void executeSafely(Runnable runnable) {
*
* @implNote This is sent asynchronously via an executor to minimize blocking. Messages are sent
* serially. If we recently sent a message before we attempt to schedule the health check, the
- * stream has been restarted/closed, there is an active health check that hasn't completed due
- * to flow control/pushback or there was a more recent send by the time we enter the
- * synchronized block, we skip the attempt to send scheduled the health check.
+ * stream has been restarted/closed, there is a scheduled health check that hasn't completed
+ * or there was a more recent send by the time we enter the synchronized block, we skip the
+ * attempt to send scheduled the health check.
*/
- public final void maybeSendHealthCheck(Instant lastSendThreshold) {
- if (debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis() && !isHealthCheckActive) {
+ public final void maybeScheduleHealthCheck(Instant lastSendThreshold) {
+ if (debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()
+ && isHealthCheckScheduled.compareAndSet(false, true)) {
// Don't block other streams when sending health check.
executeSafely(
() -> {
synchronized (this) {
- if (!isHealthCheckActive
- && !clientClosed
+ if (!clientClosed
&& debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()) {
- isHealthCheckActive = true;
try {
sendHealthCheck();
} catch (Exception e) {
logger.debug("Received exception sending health check.", e);
}
- isHealthCheckActive = false;
}
+
+ // Ready to send another health check after we attempt the scheduled health check.
Review Comment:
done
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java:
##########
@@ -244,27 +244,28 @@ protected final void executeSafely(Runnable runnable) {
*
* @implNote This is sent asynchronously via an executor to minimize blocking. Messages are sent
* serially. If we recently sent a message before we attempt to schedule the health check, the
- * stream has been restarted/closed, there is an active health check that hasn't completed due
- * to flow control/pushback or there was a more recent send by the time we enter the
- * synchronized block, we skip the attempt to send scheduled the health check.
+ * stream has been restarted/closed, there is a scheduled health check that hasn't completed
+ * or there was a more recent send by the time we enter the synchronized block, we skip the
+ * attempt to send scheduled the health check.
*/
- public final void maybeSendHealthCheck(Instant lastSendThreshold) {
- if (debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis() && !isHealthCheckActive) {
+ public final void maybeScheduleHealthCheck(Instant lastSendThreshold) {
+ if (debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()
+ && isHealthCheckScheduled.compareAndSet(false, true)) {
// Don't block other streams when sending health check.
executeSafely(
() -> {
synchronized (this) {
- if (!isHealthCheckActive
- && !clientClosed
+ if (!clientClosed
&& debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()) {
- isHealthCheckActive = true;
try {
sendHealthCheck();
} catch (Exception e) {
logger.debug("Received exception sending health check.", e);
}
- isHealthCheckActive = false;
}
+
+ // Ready to send another health check after we attempt the scheduled health check.
+ isHealthCheckScheduled.set(false);
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2016118734
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java:
##########
@@ -244,27 +244,28 @@ protected final void executeSafely(Runnable runnable) {
*
* @implNote This is sent asynchronously via an executor to minimize blocking. Messages are sent
* serially. If we recently sent a message before we attempt to schedule the health check, the
- * stream has been restarted/closed, there is an active health check that hasn't completed due
- * to flow control/pushback or there was a more recent send by the time we enter the
- * synchronized block, we skip the attempt to send scheduled the health check.
+ * stream has been restarted/closed, there is a scheduled health check that hasn't completed
+ * or there was a more recent send by the time we enter the synchronized block, we skip the
+ * attempt to send scheduled the health check.
*/
- public final void maybeSendHealthCheck(Instant lastSendThreshold) {
- if (debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis() && !isHealthCheckActive) {
+ public final void maybeScheduleHealthCheck(Instant lastSendThreshold) {
+ if (debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()
+ && isHealthCheckScheduled.compareAndSet(false, true)) {
// Don't block other streams when sending health check.
executeSafely(
() -> {
synchronized (this) {
- if (!isHealthCheckActive
- && !clientClosed
+ if (!clientClosed
&& debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()) {
- isHealthCheckActive = true;
try {
sendHealthCheck();
} catch (Exception e) {
logger.debug("Received exception sending health check.", e);
}
- isHealthCheckActive = false;
}
+
+ // Ready to send another health check after we attempt the scheduled health check.
+ isHealthCheckScheduled.set(false);
Review Comment:
how about showing if health check is scheduled in the appendSummaryHtml?
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -84,68 +85,60 @@ public void testShutdown_notBlockedBySend() throws InterruptedException, Executi
}
@Test
- public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
- TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ public void testMaybeScheduleHealthCheck() {
+ TestCallStreamObserver callStreamObserver =
+ new TestCallStreamObserver(/* waitForSend= */ false);
Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
ignored -> callStreamObserver;
TestStream testStream = newStream(clientFactory);
testStream.start();
- ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
- Future<?> sendFuture =
- sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
-
- // Sleep a bit to give sendExecutor time to execute the send().
- Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
-
- callStreamObserver.unblockSend();
-
- // Make sure the future completes.
- sendFuture.get();
+ testStream.maybeScheduleHealthCheck(reportingThreshold);
+ testStream.waitForHealthChecks(1);
assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
testStream.shutdown();
}
@Test
- public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
- throws ExecutionException, InterruptedException {
- TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ public void testMaybeSendHealthCheck_doesNotSendIfLastScheduleLessThanThreshold() {
+ TestCallStreamObserver callStreamObserver =
+ new TestCallStreamObserver(/* waitForSend= */ false);
Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
ignored -> callStreamObserver;
TestStream testStream = newStream(clientFactory);
testStream.start();
- ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
- Future<?> sendFuture =
- sendExecutor.submit(
- () -> {
- try {
- testStream.trySend(1);
- } catch (WindmillStreamShutdownException e) {
- throw new RuntimeException(e);
- }
-
- // Sleep a bit to give sendExecutor time to execute the send().
- Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
-
- // Set a really long reporting threshold.
- Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
- // Should not send health checks since we just sent the above message.
- testStream.maybeSendHealthCheck(reportingThreshold);
- testStream.maybeSendHealthCheck(reportingThreshold);
- });
- callStreamObserver.unblockSend();
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // Set a really long reporting threshold.
+ Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
+
+ // Should not send health checks since we just sent the above message.
+ testStream.maybeScheduleHealthCheck(reportingThreshold);
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ testStream.maybeScheduleHealthCheck(reportingThreshold);
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.waitForSend();
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
Review Comment:
// Sleep just to ensure a aysnc health check doesn't show up
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -197,18 +202,37 @@ protected void shutdownInternal() {}
}
private static class TestCallStreamObserver extends CallStreamObserver<Integer> {
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractWindmillStreamTest.class);
private final CountDownLatch sendBlocker = new CountDownLatch(1);
+ private final boolean waitForSend;
+
+ private TestCallStreamObserver(boolean waitForSend) {
+ this.waitForSend = waitForSend;
+ }
+
private void unblockSend() {
sendBlocker.countDown();
}
- @Override
- public void onNext(Integer integer) {
+ private void waitForSend() {
try {
- sendBlocker.await();
+ int waitedMillis = 0;
+ while (!sendBlocker.await(100, TimeUnit.MILLISECONDS)) {
+ waitedMillis += 100;
+ LOG.info("Waiting from send for {}ms", waitedMillis);
Review Comment:
I think sendBlocker is being used two different ways and is a little confusing.
How about naming this method to waitForSendUnblocked and logging here that you are waiting for send to be unblocked.
And then below in onNext instead of decrementing sendblocker, instead you increment a count of number of received and can add a new method to wait for N messages to be sent.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java:
##########
@@ -244,27 +244,28 @@ protected final void executeSafely(Runnable runnable) {
*
* @implNote This is sent asynchronously via an executor to minimize blocking. Messages are sent
* serially. If we recently sent a message before we attempt to schedule the health check, the
- * stream has been restarted/closed, there is an active health check that hasn't completed due
- * to flow control/pushback or there was a more recent send by the time we enter the
- * synchronized block, we skip the attempt to send scheduled the health check.
+ * stream has been restarted/closed, there is a scheduled health check that hasn't completed
+ * or there was a more recent send by the time we enter the synchronized block, we skip the
+ * attempt to send scheduled the health check.
*/
- public final void maybeSendHealthCheck(Instant lastSendThreshold) {
- if (debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis() && !isHealthCheckActive) {
+ public final void maybeScheduleHealthCheck(Instant lastSendThreshold) {
+ if (debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()
+ && isHealthCheckScheduled.compareAndSet(false, true)) {
// Don't block other streams when sending health check.
executeSafely(
() -> {
synchronized (this) {
- if (!isHealthCheckActive
- && !clientClosed
+ if (!clientClosed
&& debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()) {
- isHealthCheckActive = true;
try {
sendHealthCheck();
} catch (Exception e) {
logger.debug("Received exception sending health check.", e);
}
- isHealthCheckActive = false;
}
+
+ // Ready to send another health check after we attempt the scheduled health check.
Review Comment:
seems like it would be a little safer to move the try to cover all of this lambda
so that we are sure to set it back to false.
try {
sycnrhonized () { ... maybe send }
} catch {
log
} finally {
isHealthCheckScheduled.set(false);
}
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -84,68 +85,60 @@ public void testShutdown_notBlockedBySend() throws InterruptedException, Executi
}
@Test
- public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
- TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ public void testMaybeScheduleHealthCheck() {
+ TestCallStreamObserver callStreamObserver =
+ new TestCallStreamObserver(/* waitForSend= */ false);
Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
ignored -> callStreamObserver;
TestStream testStream = newStream(clientFactory);
testStream.start();
- ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
- Future<?> sendFuture =
- sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
-
- // Sleep a bit to give sendExecutor time to execute the send().
- Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
-
- callStreamObserver.unblockSend();
-
- // Make sure the future completes.
- sendFuture.get();
+ testStream.maybeScheduleHealthCheck(reportingThreshold);
+ testStream.waitForHealthChecks(1);
assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
testStream.shutdown();
}
@Test
- public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
- throws ExecutionException, InterruptedException {
- TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ public void testMaybeSendHealthCheck_doesNotSendIfLastScheduleLessThanThreshold() {
+ TestCallStreamObserver callStreamObserver =
+ new TestCallStreamObserver(/* waitForSend= */ false);
Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
ignored -> callStreamObserver;
TestStream testStream = newStream(clientFactory);
testStream.start();
- ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
- Future<?> sendFuture =
- sendExecutor.submit(
- () -> {
- try {
- testStream.trySend(1);
- } catch (WindmillStreamShutdownException e) {
- throw new RuntimeException(e);
- }
-
- // Sleep a bit to give sendExecutor time to execute the send().
- Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
-
- // Set a really long reporting threshold.
- Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
- // Should not send health checks since we just sent the above message.
- testStream.maybeSendHealthCheck(reportingThreshold);
- testStream.maybeSendHealthCheck(reportingThreshold);
- });
- callStreamObserver.unblockSend();
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
Review Comment:
doesn't trySend send inline? can this be removed?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2025843072
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +80,65 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Review Comment:
good catch here, so it seems like `send()` and `shutdown()` won't block each other IF they are not blocked due to the `CountDownLatch.await(....)`, but since this test blocks in `onNext()`it's not working, we should use onReady to do the blocking
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2025004231
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -148,19 +171,101 @@ protected boolean hasPendingRequests() {
@Override
protected void startThrottleTimer() {}
- public void testSend(Integer i)
- throws ResettableThrowingStreamObserver.StreamClosedException,
- WindmillStreamShutdownException {
+ public void testSend(Integer i) throws WindmillStreamShutdownException {
trySend(i);
}
@Override
- protected void sendHealthCheck() {}
+ protected void sendHealthCheck() {
+ numHealthChecks.incrementAndGet();
+ }
+
+ private void waitForHealthChecks(int expectedHealthChecks) {
+ int waitedMillis = 0;
+ while (numHealthChecks.get() < expectedHealthChecks) {
+ LOG.info(
+ "Waited for {}ms for {} health checks. Current health check count is {}.",
+ waitedMillis,
+ numHealthChecks.get(),
+ expectedHealthChecks);
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ }
+ }
@Override
protected void appendSpecificHtml(PrintWriter writer) {}
@Override
protected void shutdownInternal() {}
}
+
+ private static class TestCallStreamObserver extends CallStreamObserver<Integer> {
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractWindmillStreamTest.class);
+ private final CountDownLatch sendBlocker = new CountDownLatch(1);
+ private final AtomicInteger numSends = new AtomicInteger();
+
+ private final boolean waitForSend;
+
+ private TestCallStreamObserver(boolean waitForSend) {
Review Comment:
can this param and the member be called blockSend?
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +80,65 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Review Comment:
seems like this should be before the shutdown() if we're tryign to give send a chance to be blocked before shutting down.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2025841191
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -148,19 +171,101 @@ protected boolean hasPendingRequests() {
@Override
protected void startThrottleTimer() {}
- public void testSend(Integer i)
- throws ResettableThrowingStreamObserver.StreamClosedException,
- WindmillStreamShutdownException {
+ public void testSend(Integer i) throws WindmillStreamShutdownException {
trySend(i);
}
@Override
- protected void sendHealthCheck() {}
+ protected void sendHealthCheck() {
+ numHealthChecks.incrementAndGet();
+ }
+
+ private void waitForHealthChecks(int expectedHealthChecks) {
+ int waitedMillis = 0;
+ while (numHealthChecks.get() < expectedHealthChecks) {
+ LOG.info(
+ "Waited for {}ms for {} health checks. Current health check count is {}.",
+ waitedMillis,
+ numHealthChecks.get(),
+ expectedHealthChecks);
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ }
+ }
@Override
protected void appendSpecificHtml(PrintWriter writer) {}
@Override
protected void shutdownInternal() {}
}
+
+ private static class TestCallStreamObserver extends CallStreamObserver<Integer> {
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractWindmillStreamTest.class);
+ private final CountDownLatch sendBlocker = new CountDownLatch(1);
+ private final AtomicInteger numSends = new AtomicInteger();
+
+ private final boolean waitForSend;
+
+ private TestCallStreamObserver(boolean waitForSend) {
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on PR #34283:
URL: https://github.com/apache/beam/pull/34283#issuecomment-2737734854
ready for a another look @scwhittle
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on PR #34283:
URL: https://github.com/apache/beam/pull/34283#issuecomment-2755861357
ready for a another look @scwhittle
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2023487998
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -84,68 +85,60 @@ public void testShutdown_notBlockedBySend() throws InterruptedException, Executi
}
@Test
- public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
- TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ public void testMaybeScheduleHealthCheck() {
+ TestCallStreamObserver callStreamObserver =
+ new TestCallStreamObserver(/* waitForSend= */ false);
Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
ignored -> callStreamObserver;
TestStream testStream = newStream(clientFactory);
testStream.start();
- ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
- Future<?> sendFuture =
- sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
-
- // Sleep a bit to give sendExecutor time to execute the send().
- Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
-
- callStreamObserver.unblockSend();
-
- // Make sure the future completes.
- sendFuture.get();
+ testStream.maybeScheduleHealthCheck(reportingThreshold);
+ testStream.waitForHealthChecks(1);
assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
testStream.shutdown();
}
@Test
- public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
- throws ExecutionException, InterruptedException {
- TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ public void testMaybeSendHealthCheck_doesNotSendIfLastScheduleLessThanThreshold() {
+ TestCallStreamObserver callStreamObserver =
+ new TestCallStreamObserver(/* waitForSend= */ false);
Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
ignored -> callStreamObserver;
TestStream testStream = newStream(clientFactory);
testStream.start();
- ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
- Future<?> sendFuture =
- sendExecutor.submit(
- () -> {
- try {
- testStream.trySend(1);
- } catch (WindmillStreamShutdownException e) {
- throw new RuntimeException(e);
- }
-
- // Sleep a bit to give sendExecutor time to execute the send().
- Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
-
- // Set a really long reporting threshold.
- Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
- // Should not send health checks since we just sent the above message.
- testStream.maybeSendHealthCheck(reportingThreshold);
- testStream.maybeSendHealthCheck(reportingThreshold);
- });
- callStreamObserver.unblockSend();
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // Set a really long reporting threshold.
+ Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
+
+ // Should not send health checks since we just sent the above message.
+ testStream.maybeScheduleHealthCheck(reportingThreshold);
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ testStream.maybeScheduleHealthCheck(reportingThreshold);
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.waitForSend();
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2023487367
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -84,68 +85,60 @@ public void testShutdown_notBlockedBySend() throws InterruptedException, Executi
}
@Test
- public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
- TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ public void testMaybeScheduleHealthCheck() {
+ TestCallStreamObserver callStreamObserver =
+ new TestCallStreamObserver(/* waitForSend= */ false);
Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
ignored -> callStreamObserver;
TestStream testStream = newStream(clientFactory);
testStream.start();
- ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
- Future<?> sendFuture =
- sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
-
- // Sleep a bit to give sendExecutor time to execute the send().
- Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
-
- callStreamObserver.unblockSend();
-
- // Make sure the future completes.
- sendFuture.get();
+ testStream.maybeScheduleHealthCheck(reportingThreshold);
+ testStream.waitForHealthChecks(1);
assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
testStream.shutdown();
}
@Test
- public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
- throws ExecutionException, InterruptedException {
- TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ public void testMaybeSendHealthCheck_doesNotSendIfLastScheduleLessThanThreshold() {
+ TestCallStreamObserver callStreamObserver =
+ new TestCallStreamObserver(/* waitForSend= */ false);
Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
ignored -> callStreamObserver;
TestStream testStream = newStream(clientFactory);
testStream.start();
- ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
- Future<?> sendFuture =
- sendExecutor.submit(
- () -> {
- try {
- testStream.trySend(1);
- } catch (WindmillStreamShutdownException e) {
- throw new RuntimeException(e);
- }
-
- // Sleep a bit to give sendExecutor time to execute the send().
- Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
-
- // Set a really long reporting threshold.
- Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
- // Should not send health checks since we just sent the above message.
- testStream.maybeSendHealthCheck(reportingThreshold);
- testStream.maybeSendHealthCheck(reportingThreshold);
- });
- callStreamObserver.unblockSend();
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r1995469504
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java:
##########
@@ -236,14 +236,20 @@ protected final void executeSafely(Runnable runnable) {
}
}
- public final synchronized void maybeSendHealthCheck(Instant lastSendThreshold) {
- if (!clientClosed && debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()) {
- try {
- sendHealthCheck();
- } catch (Exception e) {
- logger.debug("Received exception sending health check.", e);
- }
- }
+ public final void maybeSendHealthCheck(Instant lastSendThreshold) {
Review Comment:
I think that we might want to have the check if a health-check should be scheduled before scheduling on the executor.
Otherwise there may be more churn on the executor than necessary maybe adding to contention if maybeSendHealthCheck is called more often. We probably want to keep track if there is an active health-check with a member boolean as well to avoid piling up in the executor since maybeSendHealthCheck is no longer blocking and may be called more frequently than actually sending the health checks takes.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2001984904
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java:
##########
@@ -236,14 +236,20 @@ protected final void executeSafely(Runnable runnable) {
}
}
- public final synchronized void maybeSendHealthCheck(Instant lastSendThreshold) {
- if (!clientClosed && debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()) {
- try {
- sendHealthCheck();
- } catch (Exception e) {
- logger.debug("Received exception sending health check.", e);
- }
- }
+ public final void maybeSendHealthCheck(Instant lastSendThreshold) {
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #34283:
URL: https://github.com/apache/beam/pull/34283#issuecomment-2722576566
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @johnjcasey added as fallback since no labels match configuration
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).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2011714194
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java:
##########
@@ -236,13 +239,34 @@ protected final void executeSafely(Runnable runnable) {
}
}
- public final synchronized void maybeSendHealthCheck(Instant lastSendThreshold) {
- if (!clientClosed && debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis()) {
- try {
- sendHealthCheck();
- } catch (Exception e) {
- logger.debug("Received exception sending health check.", e);
- }
+ /**
+ * Schedule an application level keep-alive health check to be sent on the stream.
+ *
+ * @implNote This is sent asynchronously via an executor to minimize blocking. Messages are sent
+ * serially. If we recently sent a message before we attempt to schedule the health check, the
+ * stream has been restarted/closed, there is an active health check that hasn't completed due
+ * to flow control/pushback or there was a more recent send by the time we enter the
+ * synchronized block, we skip the attempt to send scheduled the health check.
+ */
+ public final void maybeSendHealthCheck(Instant lastSendThreshold) {
+ if (debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis() && !isHealthCheckActive) {
Review Comment:
maybe it should be !isHealthCheckScheduled and then set it to true here before putting on the executor?
if the executor gets backlogged we could still end up queuing a lot of things on the executor since currently we don't mark active until it schedules
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
Review Comment:
I don't think you need to schedule this on a executor because maybeSendHealthCheck is no longer blocking.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
+ sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.unblockSend();
+
+ // Make sure the future completes.
+ sendFuture.get();
+
+ assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
+ testStream.shutdown();
+ }
+
+ @Test
+ public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
+ throws ExecutionException, InterruptedException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Future<?> sendFuture =
+ sendExecutor.submit(
+ () -> {
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // Set a really long reporting threshold.
+ Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
+ // Should not send health checks since we just sent the above message.
+ testStream.maybeSendHealthCheck(reportingThreshold);
Review Comment:
might need to sleep or verify the internal executor for the stream has nothing queud before you check no health-checks were sent
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
+ sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.unblockSend();
+
+ // Make sure the future completes.
+ sendFuture.get();
+
+ assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
Review Comment:
not sure if this is going to be flaky since lambda queued by trySend could possibly not have actually executed yet. Seems like it could be safer to loop while it is not yet 1.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
+ sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.unblockSend();
+
+ // Make sure the future completes.
+ sendFuture.get();
+
+ assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
+ testStream.shutdown();
+ }
+
+ @Test
+ public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
+ throws ExecutionException, InterruptedException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Future<?> sendFuture =
+ sendExecutor.submit(
+ () -> {
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // Set a really long reporting threshold.
+ Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
+ // Should not send health checks since we just sent the above message.
+ testStream.maybeSendHealthCheck(reportingThreshold);
+ testStream.maybeSendHealthCheck(reportingThreshold);
+ });
+
+ callStreamObserver.unblockSend();
Review Comment:
can we just start the test with it unblocked? Seems like you could just remove the sendFuture then and just do everything from the main test thread.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] do not block other threads from health checking if a stream is blocked [beam]
Posted by "m-trieu (via GitHub)" <gi...@apache.org>.
m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2015065774
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
+ sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.unblockSend();
+
+ // Make sure the future completes.
+ sendFuture.get();
+
+ assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
+ testStream.shutdown();
+ }
+
+ @Test
+ public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
+ throws ExecutionException, InterruptedException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Future<?> sendFuture =
+ sendExecutor.submit(
+ () -> {
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // Set a really long reporting threshold.
+ Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
+ // Should not send health checks since we just sent the above message.
+ testStream.maybeSendHealthCheck(reportingThreshold);
Review Comment:
done
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -110,12 +79,75 @@ public void setMessageCompression(boolean b) {}
// Sleep a bit to give sendExecutor time to execute the send().
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
- sendBlocker.countDown();
+ callStreamObserver.unblockSend();
assertThat(sendFuture.get()).isInstanceOf(WindmillStreamShutdownException.class);
}
+ @Test
+ public void testMaybeSendHealthCheck() throws InterruptedException, ExecutionException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+ Future<?> sendFuture =
+ sendExecutor.submit(() -> testStream.maybeSendHealthCheck(reportingThreshold));
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ callStreamObserver.unblockSend();
+
+ // Make sure the future completes.
+ sendFuture.get();
+
+ assertThat(testStream.numHealthChecks.get()).isEqualTo(1);
+ testStream.shutdown();
+ }
+
+ @Test
+ public void testMaybeSendHealthCheck_doesNotSendIfLastSendLessThanThreshold()
+ throws ExecutionException, InterruptedException {
+ TestCallStreamObserver callStreamObserver = new TestCallStreamObserver();
+ Function<StreamObserver<Integer>, StreamObserver<Integer>> clientFactory =
+ ignored -> callStreamObserver;
+
+ TestStream testStream = newStream(clientFactory);
+ testStream.start();
+ ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+ Future<?> sendFuture =
+ sendExecutor.submit(
+ () -> {
+ try {
+ testStream.trySend(1);
+ } catch (WindmillStreamShutdownException e) {
+ throw new RuntimeException(e);
+ }
+
+ // Sleep a bit to give sendExecutor time to execute the send().
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // Set a really long reporting threshold.
+ Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
+ // Should not send health checks since we just sent the above message.
+ testStream.maybeSendHealthCheck(reportingThreshold);
+ testStream.maybeSendHealthCheck(reportingThreshold);
+ });
+
+ callStreamObserver.unblockSend();
Review Comment:
done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org