flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject flink git commit: [FLINK-2528] [tests] Increase robustness and error reporting of MatchTaskTest
Date Tue, 25 Aug 2015 14:06:16 GMT
Repository: flink
Updated Branches:
  refs/heads/master 5156a1b3f -> 6f07c5f3a


[FLINK-2528] [tests] Increase robustness and error reporting of MatchTaskTest


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6f07c5f3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6f07c5f3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6f07c5f3

Branch: refs/heads/master
Commit: 6f07c5f3af1b4fd67cb61ad140b99095f7ee6c45
Parents: 5156a1b
Author: Stephan Ewen <sewen@apache.org>
Authored: Tue Aug 25 14:42:47 2015 +0200
Committer: Stephan Ewen <sewen@apache.org>
Committed: Tue Aug 25 14:42:47 2015 +0200

----------------------------------------------------------------------
 .../flink/runtime/operators/MatchTaskTest.java  | 280 ++++++++++---------
 .../operators/testutils/DriverTestBase.java     |  22 +-
 2 files changed, 170 insertions(+), 132 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/6f07c5f3/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskTest.java
index 8fbf05e..15f3d0c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/MatchTaskTest.java
@@ -16,12 +16,12 @@
  * limitations under the License.
  */
 
-
 package org.apache.flink.runtime.operators;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.functions.FlatJoinFunction;
@@ -38,9 +38,12 @@ import org.apache.flink.types.IntValue;
 import org.apache.flink.types.Key;
 import org.apache.flink.types.Record;
 import org.apache.flink.util.Collector;
+
 import org.junit.Assert;
 import org.junit.Test;
 
+import static org.junit.Assert.*;
+
 @SuppressWarnings("deprecation")
 public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record, Record,
Record>> {
 	
@@ -58,11 +61,11 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record,
Recor
 	
 	@SuppressWarnings("unchecked")
 	private final RecordComparator comparator1 = new RecordComparator(
-		new int[]{0}, (Class<? extends Key<?>>[])new Class[]{ IntValue.class });
+		new int[]{0}, (Class<? extends Key<?>>[])new Class<?>[]{ IntValue.class
});
 	
 	@SuppressWarnings("unchecked")
 	private final RecordComparator comparator2 = new RecordComparator(
-		new int[]{0}, (Class<? extends Key<?>>[])new Class[]{ IntValue.class });
+		new int[]{0}, (Class<? extends Key<?>>[])new Class<?>[]{ IntValue.class
});
 	
 	private final List<Record> outList = new ArrayList<Record>();
 	
@@ -393,150 +396,177 @@ public class MatchTaskTest extends DriverTestBase<FlatJoinFunction<Record,
Recor
 	
 	@Test
 	public void testCancelMatchTaskWhileSort1() {
-		int keyCnt = 20;
-		int valCnt = 20;
-		
-		setOutput(new NirvanaOutputList());
-		addDriverComparator(this.comparator1);
-		addDriverComparator(this.comparator2);
-		getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
-		getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
-		getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
-		setNumFileHandlesForSort(4);
-		
-		final JoinDriver<Record, Record, Record> testTask = new JoinDriver<Record, Record,
Record>();
+		final int keyCnt = 20;
+		final int valCnt = 20;
 		
 		try {
-			addInputSorted(new DelayingInfinitiveInputIterator(100), this.comparator1.duplicate());
-			addInput(new UniformRecordGenerator(keyCnt, valCnt, true));
-		} catch (Exception e) {
-			e.printStackTrace();
-			Assert.fail("The test caused an exception.");
-		}
-		
-		final AtomicBoolean success = new AtomicBoolean(false);
-		
-		Thread taskRunner = new Thread() {
-			@Override
-			public void run() {
-				try {
-					testDriver(testTask, MockMatchStub.class);
-					success.set(true);
-				} catch (Exception ie) {
-					ie.printStackTrace();
+			setOutput(new NirvanaOutputList());
+			addDriverComparator(this.comparator1);
+			addDriverComparator(this.comparator2);
+			getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+			getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+			getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
+			setNumFileHandlesForSort(4);
+			
+			final JoinDriver<Record, Record, Record> testTask = new JoinDriver<Record, Record,
Record>();
+			
+			try {
+				addInputSorted(new DelayingInfinitiveInputIterator(100), this.comparator1.duplicate());
+				addInput(new UniformRecordGenerator(keyCnt, valCnt, true));
+			} catch (Exception e) {
+				e.printStackTrace();
+				Assert.fail("The test caused an exception.");
+			}
+	
+			final AtomicReference<Throwable> error = new AtomicReference<>();
+
+			Thread taskRunner = new Thread("Task runner for testCancelMatchTaskWhileSort1()") {
+				@Override
+				public void run() {
+					try {
+						testDriver(testTask, MockMatchStub.class);
+					}
+					catch (Throwable t) {
+						error.set(t);
+					}
 				}
+			};
+			taskRunner.start();
+
+			Thread.sleep(1000);
+
+			cancel();
+			taskRunner.interrupt();
+
+			taskRunner.join(60000);
+
+			assertFalse("Task thread did not finish within 60 seconds", taskRunner.isAlive());
+
+			Throwable taskError = error.get();
+			if (taskError != null) {
+				taskError.printStackTrace();
+				fail("Error in task while canceling: " + taskError.getMessage());
 			}
-		};
-		taskRunner.start();
-		
-		TaskCancelThread tct = new TaskCancelThread(1, taskRunner, this);
-		tct.start();
-		
-		try {
-			tct.join();
-			taskRunner.join();
-		} catch(InterruptedException ie) {
-			Assert.fail("Joining threads failed");
 		}
-		
-		Assert.assertTrue("Test threw an exception even though it was properly canceled.", success.get());
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
 	}
 	
 	@Test
 	public void testCancelMatchTaskWhileSort2() {
-		int keyCnt = 20;
-		int valCnt = 20;
-		
-		setOutput(new NirvanaOutputList());
-		addDriverComparator(this.comparator1);
-		addDriverComparator(this.comparator2);
-		getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
-		getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
-		getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
-		setNumFileHandlesForSort(4);
-		
-		final JoinDriver<Record, Record, Record> testTask = new JoinDriver<Record, Record,
Record>();
+		final int keyCnt = 20;
+		final int valCnt = 20;
 		
 		try {
-			addInput(new UniformRecordGenerator(keyCnt, valCnt, true));
-			addInputSorted(new DelayingInfinitiveInputIterator(100), this.comparator1.duplicate());
-		} catch (Exception e) {
-			e.printStackTrace();
-			Assert.fail("The test caused an exception.");
-		}
-		
-		final AtomicBoolean success = new AtomicBoolean(false);
-		
-		Thread taskRunner = new Thread() {
-			@Override
-			public void run() {
-				try {
-					testDriver(testTask, MockMatchStub.class);
-					success.set(true);
-				} catch (Exception ie) {
-					ie.printStackTrace();
+			setOutput(new NirvanaOutputList());
+			addDriverComparator(this.comparator1);
+			addDriverComparator(this.comparator2);
+			getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+			getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+			getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
+			setNumFileHandlesForSort(4);
+			
+			final JoinDriver<Record, Record, Record> testTask = new JoinDriver<Record, Record,
Record>();
+			
+			try {
+				addInput(new UniformRecordGenerator(keyCnt, valCnt, true));
+				addInputSorted(new DelayingInfinitiveInputIterator(100), this.comparator1.duplicate());
+			} catch (Exception e) {
+				e.printStackTrace();
+				Assert.fail("The test caused an exception.");
+			}
+
+			final AtomicReference<Throwable> error = new AtomicReference<>();
+
+			Thread taskRunner = new Thread("Task runner for testCancelMatchTaskWhileSort2()") {
+				@Override
+				public void run() {
+					try {
+						testDriver(testTask, MockMatchStub.class);
+					}
+					catch (Throwable t) {
+						error.set(t);
+					}
 				}
+			};
+			taskRunner.start();
+
+			Thread.sleep(1000);
+
+			cancel();
+			taskRunner.interrupt();
+
+			taskRunner.join(60000);
+
+			assertFalse("Task thread did not finish within 60 seconds", taskRunner.isAlive());
+
+			Throwable taskError = error.get();
+			if (taskError != null) {
+				taskError.printStackTrace();
+				fail("Error in task while canceling: " + taskError.getMessage());
 			}
-		};
-		taskRunner.start();
-		
-		TaskCancelThread tct = new TaskCancelThread(1, taskRunner, this);
-		tct.start();
-		
-		try {
-			tct.join();
-			taskRunner.join();
-		} catch(InterruptedException ie) {
-			Assert.fail("Joining threads failed");
 		}
-		
-		Assert.assertTrue("Test threw an exception even though it was properly canceled.", success.get());
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
 	}
 	
 	@Test
 	public void testCancelMatchTaskWhileMatching() {
-		int keyCnt = 20;
-		int valCnt = 20;
-		
-		setOutput(new NirvanaOutputList());
-		addDriverComparator(this.comparator1);
-		addDriverComparator(this.comparator2);
-		getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
-		getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
-		getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
-		setNumFileHandlesForSort(4);
-		
-		final JoinDriver<Record, Record, Record> testTask = new JoinDriver<Record, Record,
Record>();
-		
-		addInput(new UniformRecordGenerator(keyCnt, valCnt, true));
-		addInput(new UniformRecordGenerator(keyCnt, valCnt, true));
-		
-		final AtomicBoolean success = new AtomicBoolean(false);
+		final int keyCnt = 20;
+		final int valCnt = 20;
 		
-		Thread taskRunner = new Thread() {
-			@Override
-			public void run() {
-				try {
-					testDriver(testTask, MockDelayingMatchStub.class);
-					success.set(true);
-				} catch (Exception ie) {
-					ie.printStackTrace();
+		try {
+			setOutput(new NirvanaOutputList());
+			addDriverComparator(this.comparator1);
+			addDriverComparator(this.comparator2);
+			getTaskConfig().setDriverPairComparator(RecordPairComparatorFactory.get());
+			getTaskConfig().setDriverStrategy(DriverStrategy.MERGE);
+			getTaskConfig().setRelativeMemoryDriver(bnljn_frac);
+			setNumFileHandlesForSort(4);
+			
+			final JoinDriver<Record, Record, Record> testTask = new JoinDriver<Record, Record,
Record>();
+			
+			addInput(new UniformRecordGenerator(keyCnt, valCnt, true));
+			addInput(new UniformRecordGenerator(keyCnt, valCnt, true));
+			
+			final AtomicReference<Throwable> error = new AtomicReference<>();
+			
+			Thread taskRunner = new Thread("Task runner for testCancelMatchTaskWhileMatching()") {
+				@Override
+				public void run() {
+					try {
+						testDriver(testTask, MockDelayingMatchStub.class);
+					}
+					catch (Throwable t) {
+						error.set(t);
+					}
 				}
+			};
+			taskRunner.start();
+			
+			Thread.sleep(1000);
+			
+			cancel();
+			taskRunner.interrupt();
+			
+			taskRunner.join(60000);
+			
+			assertFalse("Task thread did not finish within 60 seconds", taskRunner.isAlive());
+			
+			Throwable taskError = error.get();
+			if (taskError != null) {
+				taskError.printStackTrace();
+				fail("Error in task while canceling: " + taskError.getMessage());
 			}
-		};
-		taskRunner.start();
-		
-		TaskCancelThread tct = new TaskCancelThread(1, taskRunner, this);
-		tct.start();
-		
-		try {
-			tct.join();
-			taskRunner.join();
-		} catch(InterruptedException ie) {
-			Assert.fail("Joining threads failed");
 		}
-		
-		Assert.assertTrue("Test threw an exception even though it was properly canceled.", success.get());
+		catch (Exception e) {
+			e.printStackTrace();
+			fail(e.getMessage());
+		}
 	}
 	
 	@Test

http://git-wip-us.apache.org/repos/asf/flink/blob/6f07c5f3/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
index e4aad98..12ca909 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
@@ -18,8 +18,6 @@
 
 package org.apache.flink.runtime.operators.testutils;
 
-import java.io.FileNotFoundException;
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.LinkedList;
@@ -88,7 +86,7 @@ public class DriverTestBase<S extends Function> extends TestLogger
implements Pa
 	
 	private PactDriver<S, Record> driver;
 	
-	private volatile boolean running;
+	private volatile boolean running = true;
 
 	private ExecutionConfig executionConfig;
 	
@@ -119,7 +117,7 @@ public class DriverTestBase<S extends Function> extends TestLogger
implements Pa
 	}
 
 	@Parameterized.Parameters
-	public static Collection<Object[]> getConfigurations() throws FileNotFoundException,
IOException {
+	public static Collection<Object[]> getConfigurations() {
 
 		LinkedList<Object[]> configs = new LinkedList<Object[]>();
 
@@ -184,7 +182,6 @@ public class DriverTestBase<S extends Function> extends TestLogger
implements Pa
 		this.stub = (S)stubClass.newInstance();
 
 		// regular running logic
-		this.running = true;
 		boolean stubOpen = false;
 
 		try {
@@ -205,6 +202,10 @@ public class DriverTestBase<S extends Function> extends TestLogger
implements Pa
 				throw new Exception("The user defined 'open()' method caused an exception: " + t.getMessage(),
t);
 			}
 
+			if (!running) {
+				return;
+			}
+			
 			// run the user code
 			driver.run();
 
@@ -222,10 +223,10 @@ public class DriverTestBase<S extends Function> extends TestLogger
implements Pa
 				try {
 					FunctionUtils.closeFunction(this.stub);
 				}
-				catch (Throwable t) {}
+				catch (Throwable ignored) {}
 			}
 
-			// if resettable driver invoke treardown
+			// if resettable driver invoke tear down
 			if (this.driver instanceof ResettablePactDriver) {
 				final ResettablePactDriver<?, ?> resDriver = (ResettablePactDriver<?, ?>)
this.driver;
 				try {
@@ -269,6 +270,13 @@ public class DriverTestBase<S extends Function> extends TestLogger
implements Pa
 	
 	public void cancel() throws Exception {
 		this.running = false;
+		
+		// compensate for races, where cancel is called before the driver is set
+		// not that this is an artifact of a bad design of this test base, where the setup
+		// of the basic properties is not separated from the invocation of the execution logic

+		while (this.driver == null) {
+			Thread.sleep(200);
+		}
 		this.driver.cancel();
 	}
 


Mime
View raw message