[hotfix] Fixes unstable ContinuousFileMonitoringTest.

This closes apache#2446

Author: kl0u

flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileMonitoringTest.java

@@ -237,26 +237,36 @@ public void testFilePathFiltering() throws Exception {
 
 	@Test
 	public void testFileSplitMonitoringReprocessWithAppended() throws Exception {
-		Set<String> uniqFilesFound = new HashSet<>();
+		final Set<String> uniqFilesFound = new HashSet<>();
 
 		FileCreator fc = new FileCreator(INTERVAL, NO_OF_FILES);
 		fc.start();
 
-		TextInputFormat format = new TextInputFormat(new Path(hdfsURI));
-		format.setFilesFilter(FilePathFilter.createDefaultFilter());
-		ContinuousFileMonitoringFunction<String> monitoringFunction =
-			new ContinuousFileMonitoringFunction<>(format, hdfsURI,
-				FileProcessingMode.PROCESS_CONTINUOUSLY, 1, INTERVAL);
-
-		monitoringFunction.open(new Configuration());
-		monitoringFunction.run(new TestingSourceContext(monitoringFunction, uniqFilesFound));
+		Thread t = new Thread(new Runnable() {
+			@Override
+			public void run() {
+				TextInputFormat format = new TextInputFormat(new Path(hdfsURI));
+				format.setFilesFilter(FilePathFilter.createDefaultFilter());
+				ContinuousFileMonitoringFunction<String> monitoringFunction =
+					new ContinuousFileMonitoringFunction<>(format, hdfsURI,
+						FileProcessingMode.PROCESS_CONTINUOUSLY, 1, INTERVAL);
+
+				try {
+					monitoringFunction.open(new Configuration());
+					monitoringFunction.run(new TestingSourceContext(monitoringFunction, uniqFilesFound));
+				} catch (Exception e) {
+					// do nothing as we interrupted the thread.
+				}
+			}
+		});
+		t.start();
 
 		// wait until the sink also sees all the splits.
 		synchronized (uniqFilesFound) {
-			while (uniqFilesFound.size() < NO_OF_FILES) {
-				uniqFilesFound.wait(7 * INTERVAL);
-			}
+			uniqFilesFound.wait();
 		}
+		t.interrupt();
+		fc.join();
 
 		Assert.assertTrue(fc.getFilesCreated().size() == NO_OF_FILES);
 		Assert.assertTrue(uniqFilesFound.size() == NO_OF_FILES);
@@ -281,13 +291,15 @@ public void testFileSplitMonitoringProcessOnce() throws Exception {
 		Set<String> uniqFilesFound = new HashSet<>();
 
 		FileCreator fc = new FileCreator(INTERVAL, 1);
+		Set<org.apache.hadoop.fs.Path> filesCreated = fc.getFilesCreated();
 		fc.start();
 
 		// to make sure that at least one file is created
-		Set<org.apache.hadoop.fs.Path> filesCreated = fc.getFilesCreated();
-		synchronized (filesCreated) {
-			if (filesCreated.size() == 0) {
-				filesCreated.wait();
+		if (filesCreated.size() == 0) {
+			synchronized (filesCreated) {
+				if (filesCreated.size() == 0) {
+					filesCreated.wait();
+				}
 			}
 		}
 		Assert.assertTrue(fc.getFilesCreated().size() >= 1);
@@ -391,17 +403,17 @@ public void collect(FileInputSplit element) {
 				Assert.fail("Duplicate file: " + filePath);
 			}
 
-			filesFound.add(filePath);
-			try {
-				if (filesFound.size() == NO_OF_FILES) {
-					this.src.cancel();
-					this.src.close();
-					synchronized (filesFound) {
+			synchronized (filesFound) {
+				filesFound.add(filePath);
+				try {
+					if (filesFound.size() == NO_OF_FILES) {
+						this.src.cancel();
+						this.src.close();
 						filesFound.notifyAll();
 					}
+				} catch (Exception e) {
+					e.printStackTrace();
 				}
-			} catch (Exception e) {
-				e.printStackTrace();
 			}
 		}