[FLINK-9758] Fix ContinuousFileProcessingTest failure due to not sett…

…ing runtimeContext

This closes apache#6260

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

@@ -19,6 +19,7 @@
 package org.apache.flink.hdfstests;
 
 import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.api.common.io.FileInputFormat;
 import org.apache.flink.api.common.io.FilePathFilter;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -55,6 +56,7 @@
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.mockito.Mockito;
 
 import java.io.File;
 import java.io.FileNotFoundException;
@@ -579,8 +581,7 @@ public boolean filterPath(Path filePath) {
  });
 
  ContinuousFileMonitoringFunction<String> monitoringFunction =
- new ContinuousFileMonitoringFunction<>(format,
- FileProcessingMode.PROCESS_ONCE, 1, INTERVAL);
+ createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_ONCE);
 
  final FileVerifyingSourceContext context =
  new FileVerifyingSourceContext(new OneShotLatch(), monitoringFunction);
@@ -632,8 +633,7 @@ public void testNestedFilesProcessing() throws Exception {
  format.setNestedFileEnumeration(true);
 
  ContinuousFileMonitoringFunction<String> monitoringFunction =
- new ContinuousFileMonitoringFunction<>(format,
- FileProcessingMode.PROCESS_ONCE, 1, INTERVAL);
+ createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_ONCE);
 
  final FileVerifyingSourceContext context =
  new FileVerifyingSourceContext(new OneShotLatch(), monitoringFunction);
@@ -674,8 +674,7 @@ public void testSortingOnModTime() throws Exception {
  FileInputSplit[] splits = format.createInputSplits(1);
 
  ContinuousFileMonitoringFunction<String> monitoringFunction =
- new ContinuousFileMonitoringFunction<>(format,
- FileProcessingMode.PROCESS_ONCE, 1, INTERVAL);
+ createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_ONCE);
 
  ModTimeVerifyingSourceContext context = new ModTimeVerifyingSourceContext(modTimes);
 
@@ -708,8 +707,7 @@ public void testProcessOnce() throws Exception {
  format.setFilesFilter(FilePathFilter.createDefaultFilter());
 
  final ContinuousFileMonitoringFunction<String> monitoringFunction =
- new ContinuousFileMonitoringFunction<>(format,
- FileProcessingMode.PROCESS_ONCE, 1, INTERVAL);
+ createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_ONCE);
 
  final FileVerifyingSourceContext context = new FileVerifyingSourceContext(latch, monitoringFunction);
 
@@ -772,7 +770,7 @@ public void testFunctionRestore() throws Exception {
  TextInputFormat format = new TextInputFormat(new Path(testBasePath));
 
  final ContinuousFileMonitoringFunction<String> monitoringFunction =
- new ContinuousFileMonitoringFunction<>(format, FileProcessingMode.PROCESS_CONTINUOUSLY, 1, INTERVAL);
+ createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_CONTINUOUSLY);
 
  StreamSource<TimestampedFileInputSplit, ContinuousFileMonitoringFunction<String>> src =
  new StreamSource<>(monitoringFunction);
@@ -823,7 +821,7 @@ public void run() {
  testHarness.close();
 
  final ContinuousFileMonitoringFunction<String> monitoringFunctionCopy =
- new ContinuousFileMonitoringFunction<>(format, FileProcessingMode.PROCESS_CONTINUOUSLY, 1, INTERVAL);
+ createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_CONTINUOUSLY);
 
  StreamSource<TimestampedFileInputSplit, ContinuousFileMonitoringFunction<String>> srcCopy =
  new StreamSource<>(monitoringFunctionCopy);
@@ -857,8 +855,7 @@ public void testProcessContinuously() throws Exception {
  format.setFilesFilter(FilePathFilter.createDefaultFilter());
 
  final ContinuousFileMonitoringFunction<String> monitoringFunction =
- new ContinuousFileMonitoringFunction<>(format,
- FileProcessingMode.PROCESS_CONTINUOUSLY, 1, INTERVAL);
+ createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_CONTINUOUSLY);
 
  final int totalNoOfFilesToBeRead = NO_OF_FILES + 1; // 1 for the bootstrap + NO_OF_FILES
  final FileVerifyingSourceContext context = new FileVerifyingSourceContext(latch,
@@ -1055,4 +1052,14 @@ private static Tuple2<org.apache.hadoop.fs.Path, String> createFileAndFillWithDa
  Assert.assertTrue("No result file present", hdfs.exists(file));
  return new Tuple2<>(file, str.toString());
  }
+
+ /**
+ * Create continuous monitoring function with 1 reader-parallelism and interval: {@link #INTERVAL}.
+ */
+ private <OUT> ContinuousFileMonitoringFunction<OUT> createTestContinuousFileMonitoringFunction(FileInputFormat<OUT> format, FileProcessingMode fileProcessingMode) {
+ ContinuousFileMonitoringFunction<OUT> monitoringFunction =
+ new ContinuousFileMonitoringFunction<>(format, fileProcessingMode, 1, INTERVAL);
+ monitoringFunction.setRuntimeContext(Mockito.mock(RuntimeContext.class));
+ return monitoringFunction;
+ }
 }