est08zw commented on a change in pull request #15411:
URL: https://github.com/apache/flink/pull/15411#discussion_r605261249
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java
##########
@@ -262,4 +267,136 @@ protected ExtendedEndpoint(
return CompletableFuture.completedFuture(isRunning());
}
}
+
+ /**
+ * Tests that the {@link RpcEndpoint} can execute the runnable. See also
{@link
+ * org.apache.flink.runtime.rpc.RpcEndpoint#runAsync(Runnable)}
+ */
+ @Test
+ public void testRunAsync() throws InterruptedException,
ExecutionException, TimeoutException {
+ RpcEndpoint dummyRpcEndpoint = new RpcEndpoint(rpcService) {};
+ final OneShotLatch latch = new OneShotLatch();
+ try {
+ dummyRpcEndpoint.start();
+ dummyRpcEndpoint.runAsync(latch::trigger);
+ latch.await(TIMEOUT.getSize(), TIMEOUT.getUnit());
+ assertTrue(latch.isTriggered());
+ } finally {
+ RpcUtils.terminateRpcEndpoint(dummyRpcEndpoint, TIMEOUT);
+ }
+ }
+
+ /**
+ * Tests that the {@link RpcEndpoint} can execute the runnable with a
delay specified in Time.
+ * See also {@link
org.apache.flink.runtime.rpc.RpcEndpoint#scheduleRunAsync(Runnable, Time)}
+ */
+ @Test
+ public void testScheduleRunAsyncTime()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ RpcEndpoint dummyRpcEndpoint = new RpcEndpoint(rpcService) {};
+ OneShotLatch latch = new OneShotLatch();
+ Time delay = Time.seconds(1);
+ try {
+ dummyRpcEndpoint.start();
+ dummyRpcEndpoint.scheduleRunAsync(latch::trigger, delay);
Review comment:
use RpcEndpoint#getMainThreadExecutor() to schedule command indirectly.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
