You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2019/07/17 11:47:50 UTC
[flink] branch release-1.9 updated: [FLINK-13307][tests] Fix
SourceStreamTaskTest test instability.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.9 by this push:
new 327e3e1 [FLINK-13307][tests] Fix SourceStreamTaskTest test instability.
327e3e1 is described below
commit 327e3e15f87c18f23c17068da7203858ad01d24d
Author: Kostas Kloudas <kk...@gmail.com>
AuthorDate: Wed Jul 17 13:47:40 2019 +0200
[FLINK-13307][tests] Fix SourceStreamTaskTest test instability.
---
.../runtime/tasks/SourceStreamTaskTest.java | 22 +++++++++++++++++++---
1 file changed, 19 insertions(+), 3 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
index 9662720..30fe424 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
@@ -42,6 +42,8 @@ import org.apache.flink.util.ExceptionUtils;
import org.junit.Assert;
import org.junit.Test;
+import javax.annotation.Nonnull;
+
import java.io.IOException;
import java.io.Serializable;
import java.util.Collections;
@@ -57,6 +59,7 @@ import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;
+import static org.apache.flink.util.Preconditions.checkState;
import static org.hamcrest.core.Is.isA;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
@@ -244,13 +247,16 @@ public class SourceStreamTaskTest {
SourceStreamTask::new,
BasicTypeInfo.STRING_TYPE_INFO);
+ final CompletableFuture<Void> operatorRunningWaitingFuture = new CompletableFuture<>();
+ ExceptionThrowingSource.setIsInRunLoopFuture(operatorRunningWaitingFuture);
+
testHarness.setupOutputForSingletonOperatorChain();
StreamConfig streamConfig = testHarness.getStreamConfig();
streamConfig.setStreamOperator(new StreamSource<>(new ExceptionThrowingSource()));
streamConfig.setOperatorID(new OperatorID());
testHarness.invoke();
- ExceptionThrowingSource.isInRunLoop.get();
+ operatorRunningWaitingFuture.get();
testHarness.getTask().cancel();
Optional<ExceptionThrowingSource.TestException> testException = Optional.empty();
@@ -275,13 +281,16 @@ public class SourceStreamTaskTest {
SourceStreamTask::new,
BasicTypeInfo.STRING_TYPE_INFO);
+ final CompletableFuture<Void> operatorRunningWaitingFuture = new CompletableFuture<>();
+ ExceptionThrowingSource.setIsInRunLoopFuture(operatorRunningWaitingFuture);
+
testHarness.setupOutputForSingletonOperatorChain();
StreamConfig streamConfig = testHarness.getStreamConfig();
streamConfig.setStreamOperator(new StreamSource<>(new ExceptionThrowingSource()));
streamConfig.setOperatorID(new OperatorID());
testHarness.invoke();
- ExceptionThrowingSource.isInRunLoop.get();
+ operatorRunningWaitingFuture.get();
testHarness.getTask().finishTask();
testHarness.waitForTaskCompletion();
@@ -469,8 +478,9 @@ public class SourceStreamTaskTest {
*/
private static class ExceptionThrowingSource implements SourceFunction<String> {
+ private static volatile CompletableFuture<Void> isInRunLoop;
+
private volatile boolean running = true;
- static CompletableFuture<Void> isInRunLoop = new CompletableFuture<>();
public static class TestException extends RuntimeException {
public TestException(String message) {
@@ -478,8 +488,14 @@ public class SourceStreamTaskTest {
}
}
+ public static void setIsInRunLoopFuture(@Nonnull final CompletableFuture<Void> waitingLatch) {
+ ExceptionThrowingSource.isInRunLoop = waitingLatch;
+ }
+
@Override
public void run(SourceContext<String> ctx) throws TestException {
+ checkState(isInRunLoop != null && !isInRunLoop.isDone());
+
while (running) {
if (!isInRunLoop.isDone()) {
isInRunLoop.complete(null);