You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by go...@apache.org on 2020/11/24 02:41:52 UTC
[flink] branch release-1.11 updated: [FLINK-20277] Fix that
ContinuousFileMonitoringFunction cannot restore from failure
This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new 39aea57 [FLINK-20277] Fix that ContinuousFileMonitoringFunction cannot restore from failure
39aea57 is described below
commit 39aea577c27bbac3c1905ddc944bb606235e33d4
Author: godfrey he <go...@apache.org>
AuthorDate: Tue Nov 24 10:35:52 2020 +0800
[FLINK-20277] Fix that ContinuousFileMonitoringFunction cannot restore from failure
---
.../hdfstests/ContinuousFileProcessingTest.java | 66 +++++++++++++++++-----
.../source/ContinuousFileMonitoringFunction.java | 8 ++-
2 files changed, 59 insertions(+), 15 deletions(-)
diff --git a/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileProcessingTest.java b/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileProcessingTest.java
index 16a3745..e1ff0d1 100644
--- a/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileProcessingTest.java
+++ b/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileProcessingTest.java
@@ -67,6 +67,7 @@ import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
@@ -655,7 +656,17 @@ public class ContinuousFileProcessingTest {
}
@Test
- public void testSortingOnModTime() throws Exception {
+ public void testSortingOnModTimeWithDefaultGlobalModificationTime() throws Exception {
+ testSortingOnModTime(-1);
+ }
+
+ @Test
+ public void testSortingOnModTimeWithGivenGlobalModificationTime() throws Exception {
+ testSortingOnModTime(2);
+ }
+
+ private void testSortingOnModTime(int modTimesIdx) throws Exception {
+ Preconditions.checkArgument(modTimesIdx < NO_OF_FILES);
String testBasePath = hdfsURI + "/" + UUID.randomUUID() + "/";
final long[] modTimes = new long[NO_OF_FILES];
@@ -663,7 +674,7 @@ public class ContinuousFileProcessingTest {
for (int i = 0; i < NO_OF_FILES; i++) {
Tuple2<org.apache.hadoop.fs.Path, String> file =
- createFileAndFillWithData(testBasePath, "file", i, "This is test line.");
+ createFileAndFillWithData(testBasePath, "file", i, "This is test line.");
Thread.sleep(400);
filesCreated[i] = file.f0;
@@ -676,14 +687,24 @@ public class ContinuousFileProcessingTest {
// this is just to verify that all splits have been forwarded later.
FileInputSplit[] splits = format.createInputSplits(1);
+ final long[] finalModTimes;
+ final long globalModificationTime;
+ if (modTimesIdx < 0) {
+ globalModificationTime = Long.MIN_VALUE;
+ finalModTimes = modTimes;
+ } else {
+ globalModificationTime = modTimes[modTimesIdx];
+ finalModTimes = Arrays.copyOfRange(modTimes, modTimesIdx + 1, NO_OF_FILES);
+ }
ContinuousFileMonitoringFunction<String> monitoringFunction =
- createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_ONCE);
+ createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_ONCE, globalModificationTime);
- ModTimeVerifyingSourceContext context = new ModTimeVerifyingSourceContext(modTimes);
+ ModTimeVerifyingSourceContext context = new ModTimeVerifyingSourceContext(finalModTimes);
monitoringFunction.open(new Configuration());
monitoringFunction.run(context);
- Assert.assertEquals(splits.length, context.getCounter());
+ Assert.assertEquals(splits.length, NO_OF_FILES);
+ Assert.assertEquals(finalModTimes.length, context.getCounter());
// delete the created files.
for (int i = 0; i < NO_OF_FILES; i++) {
@@ -759,7 +780,16 @@ public class ContinuousFileProcessingTest {
}
@Test
- public void testFunctionRestore() throws Exception {
+ public void testFunctionRestoreWithDefaultGlobalModificationTime() throws Exception {
+ testFunctionRestore(Long.MIN_VALUE);
+ }
+
+ @Test
+ public void testFunctionRestoreWithGivenGlobalModificationTime() throws Exception {
+ testFunctionRestore(100);
+ }
+
+ private void testFunctionRestore(long globalModificationTime) throws Exception {
String testBasePath = hdfsURI + "/" + UUID.randomUUID() + "/";
org.apache.hadoop.fs.Path path = null;
@@ -773,13 +803,13 @@ public class ContinuousFileProcessingTest {
TextInputFormat format = new TextInputFormat(new Path(testBasePath));
final ContinuousFileMonitoringFunction<String> monitoringFunction =
- createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_CONTINUOUSLY);
+ createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_CONTINUOUSLY, globalModificationTime);
StreamSource<TimestampedFileInputSplit, ContinuousFileMonitoringFunction<String>> src =
- new StreamSource<>(monitoringFunction);
+ new StreamSource<>(monitoringFunction);
final AbstractStreamOperatorTestHarness<TimestampedFileInputSplit> testHarness =
- new AbstractStreamOperatorTestHarness<>(src, 1, 1, 0);
+ new AbstractStreamOperatorTestHarness<>(src, 1, 1, 0);
testHarness.open();
final Throwable[] error = new Throwable[1];
@@ -824,13 +854,13 @@ public class ContinuousFileProcessingTest {
testHarness.close();
final ContinuousFileMonitoringFunction<String> monitoringFunctionCopy =
- createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_CONTINUOUSLY);
+ createTestContinuousFileMonitoringFunction(format, FileProcessingMode.PROCESS_CONTINUOUSLY, globalModificationTime);
StreamSource<TimestampedFileInputSplit, ContinuousFileMonitoringFunction<String>> srcCopy =
- new StreamSource<>(monitoringFunctionCopy);
+ new StreamSource<>(monitoringFunctionCopy);
AbstractStreamOperatorTestHarness<TimestampedFileInputSplit> testHarnessCopy =
- new AbstractStreamOperatorTestHarness<>(srcCopy, 1, 1, 0);
+ new AbstractStreamOperatorTestHarness<>(srcCopy, 1, 1, 0);
testHarnessCopy.initializeState(snapshot);
testHarnessCopy.open();
@@ -1060,8 +1090,18 @@ public class ContinuousFileProcessingTest {
* Create continuous monitoring function with 1 reader-parallelism and interval: {@link #INTERVAL}.
*/
private <OUT> ContinuousFileMonitoringFunction<OUT> createTestContinuousFileMonitoringFunction(FileInputFormat<OUT> format, FileProcessingMode fileProcessingMode) {
+ return createTestContinuousFileMonitoringFunction(format, fileProcessingMode, Long.MIN_VALUE);
+ }
+
+ /**
+ * Create continuous monitoring function with 1 reader-parallelism, interval: {@link #INTERVAL} and globalModificationTime.
+ */
+ private <OUT> ContinuousFileMonitoringFunction<OUT> createTestContinuousFileMonitoringFunction(
+ FileInputFormat<OUT> format,
+ FileProcessingMode fileProcessingMode,
+ long globalModificationTime) {
ContinuousFileMonitoringFunction<OUT> monitoringFunction =
- new ContinuousFileMonitoringFunction<>(format, fileProcessingMode, 1, INTERVAL);
+ new ContinuousFileMonitoringFunction<>(format, fileProcessingMode, 1, INTERVAL, globalModificationTime);
monitoringFunction.setRuntimeContext(Mockito.mock(RuntimeContext.class));
return monitoringFunction;
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileMonitoringFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileMonitoringFunction.java
index f73c23d..24f9d38 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileMonitoringFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileMonitoringFunction.java
@@ -96,6 +96,9 @@ public class ContinuousFileMonitoringFunction<OUT>
/** The maximum file modification time seen so far. */
private volatile long globalModificationTime;
+ /** Store the initialized value of globalModificationTime. */
+ private final long initializedGlobalModificationTime;
+
private transient Object checkpointLock;
private volatile boolean isRunning = true;
@@ -134,6 +137,7 @@ public class ContinuousFileMonitoringFunction<OUT>
this.watchType = watchType;
this.readerParallelism = Math.max(readerParallelism, 1);
this.globalModificationTime = globalModificationTime;
+ this.initializedGlobalModificationTime = globalModificationTime;
}
@VisibleForTesting
@@ -168,7 +172,7 @@ public class ContinuousFileMonitoringFunction<OUT>
Preconditions.checkArgument(retrievedStates.size() <= 1,
getClass().getSimpleName() + " retrieved invalid state.");
- if (retrievedStates.size() == 1 && globalModificationTime != Long.MIN_VALUE) {
+ if (retrievedStates.size() == 1 && globalModificationTime != initializedGlobalModificationTime) {
// this is the case where we have both legacy and new state.
// The two should be mutually exclusive for the operator, thus we throw the exception.
@@ -230,7 +234,7 @@ public class ContinuousFileMonitoringFunction<OUT>
// after a failure and we managed to have a successful
// checkpoint, we will not reprocess the directory.
- if (globalModificationTime == Long.MIN_VALUE) {
+ if (globalModificationTime == initializedGlobalModificationTime) {
monitorDirAndForwardSplits(fileSystem, context);
globalModificationTime = Long.MAX_VALUE;
}