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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
