This is an automated email from the ASF dual-hosted git repository.
chungen0126 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 80be70dcdc6 HDDS-16324. XceiverClientGrpc.streamRead should wait for
flow-control readiness via onReadyHandler instead of a 10 ms poll (#11195)
80be70dcdc6 is described below
commit 80be70dcdc61c95ff2f07f4580aa061a574ab776
Author: Sergey Soldatov <[email protected]>
AuthorDate: Tue Sep 15 11:53:09 2026 -0700
HDDS-16324. XceiverClientGrpc.streamRead should wait for flow-control
readiness via onReadyHandler instead of a 10 ms poll (#11195)
---
.../apache/hadoop/hdds/scm/XceiverClientGrpc.java | 68 ++++++--
.../hadoop/hdds/scm/StreamingReadResponse.java | 52 +++++++
.../hadoop/hdds/scm/TestStreamingReadResponse.java | 172 +++++++++++++++++++++
.../hadoop/hdds/scm/TestXceiverClientGrpc.java | 73 +++++++--
4 files changed, 334 insertions(+), 31 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
index 93e7c005934..4fe56e6ac09 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java
@@ -39,7 +39,6 @@
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.LockSupport;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.HddsUtils;
import org.apache.hadoop.hdds.client.BlockID;
@@ -72,6 +71,7 @@
import org.apache.ratis.thirdparty.io.grpc.netty.GrpcSslContexts;
import org.apache.ratis.thirdparty.io.grpc.netty.NettyChannelBuilder;
import org.apache.ratis.thirdparty.io.grpc.stub.ClientCallStreamObserver;
+import org.apache.ratis.thirdparty.io.grpc.stub.ClientResponseObserver;
import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
import org.apache.ratis.thirdparty.io.netty.handler.ssl.SslContextBuilder;
import org.slf4j.Logger;
@@ -573,16 +573,15 @@ public void streamRead(ContainerCommandRequestProto
request,
request.getReadBlock().getBlockID().getLocalID(),
request.getReadBlock().getOffset(),
request.getReadBlock().getLength());
- final long deadlineNs = System.nanoTime() + streamReadTimeoutNanos;
- while (!obs.isReady() && System.nanoTime() - deadlineNs < 0) {
- LockSupport.parkNanos(10_000_000L);
- if (Thread.currentThread().isInterrupted()) {
- Thread.currentThread().interrupt();
- throw new InterruptedIOException("Interrupted while waiting for
stream to become ready: " + streamObserver);
+ try {
+ if (!streamObserver.awaitReady(streamReadTimeoutNanos)) {
+ throw new TimeoutIOException("Timed out waiting for stream to become
ready: " + streamObserver);
}
- }
- if (!obs.isReady()) {
- throw new TimeoutIOException("Timed out waiting for stream to become
ready: " + streamObserver);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw (IOException) new InterruptedIOException(
+ "Interrupted while waiting for stream to become ready: " +
streamObserver)
+ .initCause(e);
}
}
@@ -619,11 +618,8 @@ public void initStreamRead(BlockID blockID,
StreamingReaderSpi streamObserver,
throw new IOException("Failed to get gRPC stub for DataNode: " + dn);
}
LOG.debug("initStreamRead {} on datanode {}",
blockID.getContainerBlockID(), dn);
- StreamObserver<ContainerCommandRequestProto> requestObserver = stub
- .withDeadlineAfter(timeout, TimeUnit.SECONDS)
- .send(streamObserver);
- streamObserver.setStreamingReadResponse(new StreamingReadResponse(dn,
- (ClientCallStreamObserver<ContainerCommandRequestProto>)
requestObserver));
+ stub.withDeadlineAfter(timeout, TimeUnit.SECONDS)
+ .send(new ReadyAwareResponseObserver(dn, streamObserver));
return;
} catch (IOException e) {
LOG.error("Failed to start streaming read to DataNode {}", dn, e);
@@ -641,6 +637,48 @@ public void initStreamRead(BlockID blockID,
StreamingReaderSpi streamObserver,
}
}
+ /**
+ * Registers the request stream's {@code onReadyHandler} in {@code
beforeStart}, the only point gRPC allows it,
+ * hands the {@link StreamingReadResponse} to the reader from there, and
tells it when the call terminates so a
+ * sender blocked in {@link StreamingReadResponse#awaitReady(long)} fails
right away.
+ */
+ @VisibleForTesting
+ static final class ReadyAwareResponseObserver
+ implements ClientResponseObserver<ContainerCommandRequestProto,
ContainerCommandResponseProto> {
+ private final DatanodeDetails dn;
+ private final StreamingReaderSpi reader;
+ private StreamingReadResponse response;
+
+ ReadyAwareResponseObserver(DatanodeDetails dn, StreamingReaderSpi reader) {
+ this.dn = dn;
+ this.reader = reader;
+ }
+
+ @Override
+ public void
beforeStart(ClientCallStreamObserver<ContainerCommandRequestProto>
requestStream) {
+ response = new StreamingReadResponse(dn, requestStream);
+ requestStream.setOnReadyHandler(response::signalReady);
+ reader.setStreamingReadResponse(response);
+ }
+
+ @Override
+ public void onNext(ContainerCommandResponseProto value) {
+ reader.onNext(value);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ response.signalTerminated(t);
+ reader.onError(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ response.signalTerminated(null);
+ reader.onCompleted();
+ }
+ }
+
/**
* This method should be called to indicate the end of streaming read. Its
primary purpose is to release the
* semaphore acquired when starting the streaming read, but is also used to
update any metrics or debug logs as
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/StreamingReadResponse.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/StreamingReadResponse.java
index 3018fda7ea6..e78eab4e365 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/StreamingReadResponse.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/StreamingReadResponse.java
@@ -17,6 +17,8 @@
package org.apache.hadoop.hdds.scm;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
import org.apache.ratis.thirdparty.io.grpc.stub.ClientCallStreamObserver;
@@ -27,6 +29,11 @@
*/
public class StreamingReadResponse {
+ private final Object readyLock = new Object();
+ /** Set once the call has terminated; guarded by {@link #readyLock}. */
+ private boolean terminated;
+ /** The failure that terminated the call, or null if it completed normally;
guarded by {@link #readyLock}. */
+ private Throwable terminationCause;
private final DatanodeDetails dn;
private final
ClientCallStreamObserver<ContainerProtos.ContainerCommandRequestProto>
requestObserver;
private final String name;
@@ -48,6 +55,51 @@ public
ClientCallStreamObserver<ContainerProtos.ContainerCommandRequestProto> ge
return requestObserver;
}
+ /** Registered as the request stream's {@code onReadyHandler}. */
+ public void signalReady() {
+ synchronized (readyLock) {
+ readyLock.notifyAll();
+ }
+ }
+
+ /**
+ * Called when the call has terminated, so that {@link #awaitReady(long)}
fails immediately instead of waiting
+ * for the timeout: a terminated stream never becomes ready and gRPC no
longer invokes the {@code onReadyHandler}.
+ *
+ * @param cause the error that terminated the call, or null if it completed
normally.
+ */
+ public void signalTerminated(Throwable cause) {
+ synchronized (readyLock) {
+ terminated = true;
+ terminationCause = cause;
+ readyLock.notifyAll();
+ }
+ }
+
+ /**
+ * Wait until the request stream can accept another message.
+ *
+ * @return true if the stream is ready, false if the timeout expired first.
+ * @throws IOException if the call terminated before the stream became ready.
+ */
+ public boolean awaitReady(long timeoutNanos) throws InterruptedException,
IOException {
+ final long deadlineNanos = System.nanoTime() + timeoutNanos;
+ synchronized (readyLock) {
+ while (!requestObserver.isReady()) {
+ if (terminated) {
+ throw new IOException("Stream " + name + " terminated while waiting
for it to become ready",
+ terminationCause);
+ }
+ final long remainingNanos = deadlineNanos - System.nanoTime();
+ if (remainingNanos <= 0) {
+ return false;
+ }
+ TimeUnit.NANOSECONDS.timedWait(readyLock, remainingNanos);
+ }
+ }
+ return true;
+ }
+
@Override
public String toString() {
return name;
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/TestStreamingReadResponse.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/TestStreamingReadResponse.java
new file mode 100644
index 00000000000..14a76b35e9f
--- /dev/null
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/TestStreamingReadResponse.java
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
+import
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
+import org.apache.ozone.test.GenericTestUtils;
+import org.apache.ratis.thirdparty.io.grpc.stub.ClientCallStreamObserver;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link StreamingReadResponse#awaitReady(long)}.
+ */
+class TestStreamingReadResponse {
+
+ private static final class FakeRequestObserver extends
ClientCallStreamObserver<ContainerCommandRequestProto> {
+ private final AtomicBoolean ready = new AtomicBoolean();
+
+ @Override
+ public boolean isReady() {
+ return ready.get();
+ }
+
+ @Override
+ public void setOnReadyHandler(Runnable onReadyHandler) {
+ }
+
+ @Override
+ public void disableAutoInboundFlowControl() {
+ }
+
+ @Override
+ public void request(int count) {
+ }
+
+ @Override
+ public void setMessageCompression(boolean enable) {
+ }
+
+ @Override
+ public void cancel(String message, Throwable cause) {
+ }
+
+ @Override
+ public void onNext(ContainerCommandRequestProto value) {
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+ }
+
+ private static StreamingReadResponse newResponse(FakeRequestObserver
observer) {
+ return new
StreamingReadResponse(MockDatanodeDetails.randomDatanodeDetails(), observer);
+ }
+
+ @Test
+ void returnsImmediatelyWhenAlreadyReady() throws Exception {
+ final FakeRequestObserver observer = new FakeRequestObserver();
+ observer.ready.set(true);
+ assertTrue(newResponse(observer).awaitReady(TimeUnit.SECONDS.toNanos(10)));
+ }
+
+ @Test
+ void timesOutWithoutASignal() throws Exception {
+ final FakeRequestObserver observer = new FakeRequestObserver();
+ final long start = System.nanoTime();
+
assertFalse(newResponse(observer).awaitReady(TimeUnit.MILLISECONDS.toNanos(50)));
+ assertTrue(System.nanoTime() - start >= TimeUnit.MILLISECONDS.toNanos(50),
"must wait out the timeout");
+ }
+
+ @Test
+ void wakesOnSignalReady() throws Exception {
+ final FakeRequestObserver observer = new FakeRequestObserver();
+ final StreamingReadResponse response = newResponse(observer);
+ final AtomicBoolean result = new AtomicBoolean();
+ final Thread waiter = new Thread(() -> {
+ try {
+ result.set(response.awaitReady(TimeUnit.SECONDS.toNanos(30)));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ });
+ waiter.start();
+ awaitBlocked(waiter);
+
+ // a spurious signalReady() while isReady() is still false must not wake
the waiter with a wrong result
+ response.signalReady();
+ awaitBlocked(waiter);
+ assertTrue(waiter.isAlive(), "waiter must still be waiting after a
spurious signalReady()");
+
+ observer.ready.set(true);
+ response.signalReady();
+ waiter.join(TimeUnit.SECONDS.toMillis(10));
+ assertFalse(waiter.isAlive(), "waiter did not return");
+ assertTrue(result.get());
+ }
+
+ /** Waits until {@code thread} is actually parked in {@link
StreamingReadResponse#awaitReady(long)}. */
+ private static void awaitBlocked(Thread thread) throws Exception {
+ GenericTestUtils.waitFor(() -> thread.getState() ==
Thread.State.TIMED_WAITING, 10, 10_000);
+ }
+
+ @Test
+ void failsImmediatelyWhenAlreadyTerminated() {
+ final FakeRequestObserver observer = new FakeRequestObserver();
+ final StreamingReadResponse response = newResponse(observer);
+ final Throwable cause = new RuntimeException("boom");
+ response.signalTerminated(cause);
+ final long start = System.nanoTime();
+ final IOException e = assertThrows(IOException.class, () ->
response.awaitReady(TimeUnit.SECONDS.toNanos(30)));
+ assertSame(cause, e.getCause());
+ assertTrue(System.nanoTime() - start < TimeUnit.SECONDS.toNanos(5), "must
not wait out the timeout");
+ }
+
+ @Test
+ void wakesOnSignalTerminated() throws Exception {
+ final FakeRequestObserver observer = new FakeRequestObserver();
+ final StreamingReadResponse response = newResponse(observer);
+ final CountDownLatch waiting = new CountDownLatch(1);
+ final AtomicReference<Throwable> failure = new AtomicReference<>();
+ final Thread waiter = new Thread(() -> {
+ try {
+ waiting.countDown();
+ response.awaitReady(TimeUnit.SECONDS.toNanos(30));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (IOException e) {
+ failure.set(e);
+ }
+ });
+ waiter.start();
+ assertTrue(waiting.await(10, TimeUnit.SECONDS));
+
+ response.signalTerminated(null);
+ waiter.join(TimeUnit.SECONDS.toMillis(10));
+ assertFalse(waiter.isAlive(), "waiter did not return");
+ assertTrue(failure.get() instanceof IOException, "waiter was not failed by
signalTerminated");
+ }
+}
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientGrpc.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientGrpc.java
index 8cfcac04334..b2ff5ea70d9 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientGrpc.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientGrpc.java
@@ -21,6 +21,8 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import java.io.IOException;
import java.io.InterruptedIOException;
@@ -29,6 +31,9 @@
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.lang3.RandomUtils;
import org.apache.hadoop.hdds.client.BlockID;
@@ -44,10 +49,12 @@
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
import org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls;
import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.apache.ozone.test.GenericTestUtils;
import org.apache.ratis.protocol.exceptions.TimeoutIOException;
import org.apache.ratis.thirdparty.io.grpc.stub.ClientCallStreamObserver;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
/**
* Tests for TestXceiverClientGrpc, to ensure topology aware reads work
@@ -300,7 +307,7 @@ private void
invokeXceiverClientReadSmallFile(XceiverClientSpi client)
/** streamRead() calls onNext() immediately when isReady() is true from the
start. */
@Test
public void testStreamReadSendsImmediatelyWhenReady() throws Exception {
- TrackingStreamObserver obs = new TrackingStreamObserver(0);
+ TrackingStreamObserver obs = new TrackingStreamObserver(true);
StreamingReadResponse response = new StreamingReadResponse(
MockDatanodeDetails.randomDatanodeDetails(), obs);
ContainerProtos.ContainerCommandRequestProto request =
buildReadBlockRequest();
@@ -315,21 +322,41 @@ public void testStreamReadSendsImmediatelyWhenReady()
throws Exception {
"isReady() must be checked exactly once when stream is immediately
ready");
}
- /** streamRead() spin-waits until isReady() becomes true, then calls
onNext(). */
+ /** streamRead() blocks until the onReadyHandler fires and isReady() becomes
true, then calls onNext(). */
@Test
public void testStreamReadWaitsUntilReadyThenSends() throws Exception {
- TrackingStreamObserver obs = new TrackingStreamObserver(3);
- StreamingReadResponse response = new StreamingReadResponse(
- MockDatanodeDetails.randomDatanodeDetails(), obs);
+ TrackingStreamObserver obs = new TrackingStreamObserver(false);
+ StreamingReaderSpi reader = mock(StreamingReaderSpi.class);
+ new
XceiverClientGrpc.ReadyAwareResponseObserver(MockDatanodeDetails.randomDatanodeDetails(),
reader)
+ .beforeStart(obs);
+ ArgumentCaptor<StreamingReadResponse> captured =
ArgumentCaptor.forClass(StreamingReadResponse.class);
+ verify(reader).setStreamingReadResponse(captured.capture());
ContainerProtos.ContainerCommandRequestProto request =
buildReadBlockRequest();
- try (XceiverClientGrpc client = new XceiverClientGrpc(pipeline, conf)) {
- client.streamRead(request, response);
+ OzoneConfiguration longTimeout = new OzoneConfiguration();
+ longTimeout.set("ozone.client.stream.read.timeout", "60s");
+ try (XceiverClientGrpc client = new XceiverClientGrpc(pipeline,
longTimeout)) {
+ FutureTask<Void> send = new FutureTask<>(() -> {
+ client.streamRead(request, captured.getValue());
+ return null;
+ });
+ Thread sender = new Thread(send);
+ sender.start();
+ try {
+ GenericTestUtils.waitFor(() -> sender.getState() ==
Thread.State.TIMED_WAITING, 10, 5_000);
+ assertThrows(TimeoutException.class, () -> send.get(200,
TimeUnit.MILLISECONDS));
+ assertThat(obs.getSent()).isEmpty();
+ obs.markReady();
+ send.get(5, TimeUnit.SECONDS);
+ assertThat(obs.getReadyCalls().get()).as("isReady() calls must stay
bounded while waiting").isBetween(3, 5);
+ } finally {
+ sender.interrupt();
+ sender.join(TimeUnit.SECONDS.toMillis(5));
+ }
}
assertEquals(1, obs.getSent().size(), "onNext must be called exactly
once");
assertEquals(request, obs.getSent().get(0));
- assertThat(obs.getReadyCalls().get()).isGreaterThanOrEqualTo(4);
}
/**
@@ -340,7 +367,7 @@ public void testStreamReadFailsAfterTimeoutIfNeverReady()
throws Exception {
OzoneConfiguration timeoutConf = new OzoneConfiguration();
timeoutConf.set("ozone.client.stream.read.timeout", "1s");
- TrackingStreamObserver obs = new TrackingStreamObserver(Integer.MAX_VALUE);
+ TrackingStreamObserver obs = new TrackingStreamObserver(false);
StreamingReadResponse response = new StreamingReadResponse(
MockDatanodeDetails.randomDatanodeDetails(), obs);
ContainerProtos.ContainerCommandRequestProto request =
buildReadBlockRequest();
@@ -357,10 +384,10 @@ public void testStreamReadFailsAfterTimeoutIfNeverReady()
throws Exception {
assertThat(elapsed).isLessThan(10_000L);
}
- /** streamRead() exits the spin-wait immediately on interrupt and restores
the interrupt flag. */
+ /** streamRead() exits the ready wait immediately on interrupt and restores
the interrupt flag. */
@Test
public void testStreamReadRestoresInterruptFlagOnInterruption() throws
Exception {
- TrackingStreamObserver obs = new TrackingStreamObserver(Integer.MAX_VALUE);
+ TrackingStreamObserver obs = new TrackingStreamObserver(false);
StreamingReadResponse response = new StreamingReadResponse(
MockDatanodeDetails.randomDatanodeDetails(), obs);
ContainerProtos.ContainerCommandRequestProto request =
buildReadBlockRequest();
@@ -390,16 +417,28 @@ public void
testStreamReadRestoresInterruptFlagOnInterruption() throws Exception
}
}
- /** Records onNext() calls and controls when isReady() starts returning
true. */
+ /**
+ * Records onNext() calls and models gRPC flow control: isReady() reflects a
flag that {@link #markReady()} flips,
+ * after which the registered onReadyHandler is invoked, as gRPC does when
the transport becomes writable.
+ */
private static final class TrackingStreamObserver
extends
ClientCallStreamObserver<ContainerProtos.ContainerCommandRequestProto> {
private final List<ContainerProtos.ContainerCommandRequestProto> sent =
new ArrayList<>();
private final AtomicInteger readyCalls = new AtomicInteger();
- private final int readyAfter;
+ private volatile boolean ready;
+ private volatile Runnable onReadyHandler;
- TrackingStreamObserver(int readyAfter) {
- this.readyAfter = readyAfter;
+ TrackingStreamObserver(boolean ready) {
+ this.ready = ready;
+ }
+
+ void markReady() {
+ ready = true;
+ final Runnable handler = onReadyHandler;
+ if (handler != null) {
+ handler.run();
+ }
}
List<ContainerProtos.ContainerCommandRequestProto> getSent() {
@@ -412,7 +451,8 @@ AtomicInteger getReadyCalls() {
@Override
public boolean isReady() {
- return readyCalls.incrementAndGet() > readyAfter;
+ readyCalls.incrementAndGet();
+ return ready;
}
@Override
@@ -426,6 +466,7 @@ public void cancel(String msg, Throwable cause) {
@Override
public void setOnReadyHandler(Runnable r) {
+ onReadyHandler = r;
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
