Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/2797#discussion_r88019498 --- Diff: flink-streaming-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSinkTest.java --- @@ -118,48 +146,290 @@ public static void destroyHDFS() { } @Test - public void testCheckpointWithoutNotify() throws Exception { - File dataDir = tempFolder.newFolder(); + public void testInactivityPeriodWithLateNotify() throws Exception { + final File outDir = tempFolder.newFolder(); + + OneInputStreamOperatorTestHarness<String, Object> testHarness = createRescalingTestSink(outDir, 1, 0, 100); + testHarness.setup(); + testHarness.open(); + + testHarness.setProcessingTime(0L); + + testHarness.processElement(new StreamRecord<>("test1", 1L)); + testHarness.processElement(new StreamRecord<>("test2", 1L)); + checkFs(outDir, 2, 0 ,0, 0); + + testHarness.setProcessingTime(101L); // put some in pending + checkFs(outDir, 0, 2, 0, 0); + + testHarness.snapshot(0, 0); // put them in pending for 0 + checkFs(outDir, 0, 2, 0, 0); + + testHarness.processElement(new StreamRecord<>("test3", 1L)); + testHarness.processElement(new StreamRecord<>("test4", 1L)); - OneInputStreamOperatorTestHarness<String, Object> testHarness = createTestSink(dataDir); + testHarness.setProcessingTime(202L); // put some in pending + testHarness.snapshot(1, 0); // put them in pending for 1 + checkFs(outDir, 0, 4, 0, 0); + + testHarness.notifyOfCompletedCheckpoint(0); // put the pending for 0 to the "committed" state + checkFs(outDir, 0, 2, 2, 0); + + testHarness.notifyOfCompletedCheckpoint(1); // put the pending for 1 to the "committed" state + checkFs(outDir, 0, 0, 4, 0); + } + + @Test + public void testBucketStateTransitions() throws Exception { + final File outDir = tempFolder.newFolder(); + + OneInputStreamOperatorTestHarness<String, Object> testHarness = createRescalingTestSink(outDir, 1, 0, 100); testHarness.setup(); testHarness.open(); - testHarness.processElement(new StreamRecord<>("Hello")); - testHarness.processElement(new StreamRecord<>("Hello")); - testHarness.processElement(new StreamRecord<>("Hello")); + testHarness.setProcessingTime(0L); + + testHarness.processElement(new StreamRecord<>("test1", 1L)); + testHarness.processElement(new StreamRecord<>("test2", 1L)); + checkFs(outDir, 2, 0 ,0, 0); + + // this is to check the inactivity threshold + testHarness.setProcessingTime(101L); + checkFs(outDir, 0, 2, 0, 0); + + testHarness.processElement(new StreamRecord<>("test3", 1L)); + checkFs(outDir, 1, 2, 0, 0); + + testHarness.snapshot(0, 0); + checkFs(outDir, 1, 2, 0, 0); + + testHarness.notifyOfCompletedCheckpoint(0); + checkFs(outDir, 1, 0, 2, 0); - testHarness.setProcessingTime(10000L); + OperatorStateHandles snapshot = testHarness.snapshot(1, 0); - // snapshot but don't call notify to simulate a notify that never - // arrives, the sink should move pending files in restore() in that case - StreamStateHandle snapshot1 = testHarness.snapshotLegacy(0, 0); + testHarness.close(); + checkFs(outDir, 0, 1, 2, 0); + + testHarness = createRescalingTestSink(outDir, 1, 0, 100); + testHarness.setup(); + testHarness.initializeState(snapshot); + testHarness.open(); + checkFs(outDir, 0, 0, 3, 1); + + snapshot = testHarness.snapshot(2, 0); + + testHarness.processElement(new StreamRecord<>("test4", 10)); + checkFs(outDir, 1, 0, 3, 1); - testHarness = createTestSink(dataDir); + testHarness = createRescalingTestSink(outDir, 1, 0, 100); testHarness.setup(); - testHarness.restore(snapshot1); + testHarness.initializeState(snapshot); testHarness.open(); - testHarness.processElement(new StreamRecord<>("Hello")); + // the in-progress file remains as we do not clean up now + checkFs(outDir, 1, 0, 3, 1); testHarness.close(); - int numComplete = 0; - int numPending = 0; - for (File file: FileUtils.listFiles(dataDir, null, true)) { + // at close it is not moved to final because it is not part + // of the current task's state, it was just a not cleaned up leftover. + checkFs(outDir, 1, 0, 3, 1); + } + + @Test + public void testSameParallelismWithShufflingStates() throws Exception { + final File outDir = tempFolder.newFolder(); + + OneInputStreamOperatorTestHarness<String, Object> testHarness1 = createRescalingTestSink(outDir, 2, 0, 100); + testHarness1.setup(); + testHarness1.open(); + + OneInputStreamOperatorTestHarness<String, Object> testHarness2 = createRescalingTestSink(outDir, 2, 1, 100); + testHarness2.setup(); + testHarness2.open(); + + testHarness1.processElement(new StreamRecord<>("test1", 0L)); + checkFs(outDir, 1, 0, 0, 0); + + testHarness2.processElement(new StreamRecord<>("test2", 0L)); + checkFs(outDir, 2, 0, 0, 0); + + // intentionally we snapshot them in the reverse order so that the states are shuffled + OperatorStateHandles mergedSnapshot = AbstractStreamOperatorTestHarness.repackageState( + testHarness2.snapshot(0, 0), + testHarness1.snapshot(0, 0) + ); + + checkFs(outDir, 2, 0, 0, 0); + + // this will not be included in any checkpoint so it can be cleaned up (although we do not) + testHarness2.processElement(new StreamRecord<>("test3", 0L)); + checkFs(outDir, 3, 0, 0, 0); + + testHarness1 = createRescalingTestSink(outDir, 2, 0, 100); + testHarness1.setup(); + testHarness1.initializeState(mergedSnapshot); + testHarness1.open(); + + // the one in-progress will be the one assigned to the next instance, + // the other is the test3 which is just not cleaned up + checkFs(outDir, 2, 0, 1, 1); + + testHarness2 = createRescalingTestSink(outDir, 2, 1, 100); + testHarness2.setup(); + testHarness2.initializeState(mergedSnapshot); + testHarness2.open(); + + checkFs(outDir, 1, 0, 2, 2); + + testHarness1.close(); + testHarness2.close(); + + // the 1 in-progress can be discarded. + checkFs(outDir, 1, 0, 2, 2); + } + + @Test + public void testScalingDown() throws Exception { + final File outDir = tempFolder.newFolder(); + + OneInputStreamOperatorTestHarness<String, Object> testHarness1 = createRescalingTestSink(outDir, 3, 0, 100); + testHarness1.setup(); + testHarness1.open(); + + OneInputStreamOperatorTestHarness<String, Object> testHarness2 = createRescalingTestSink(outDir, 3, 1, 100); + testHarness2.setup(); + testHarness2.open(); + + OneInputStreamOperatorTestHarness<String, Object> testHarness3 = createRescalingTestSink(outDir, 3, 2, 100); + testHarness3.setup(); + testHarness3.open(); + + testHarness1.processElement(new StreamRecord<>("test1", 0L)); + checkFs(outDir, 1, 0, 0, 0); + + testHarness2.processElement(new StreamRecord<>("test2", 0L)); + checkFs(outDir, 2, 0, 0, 0); + + testHarness3.processElement(new StreamRecord<>("test3", 0L)); + testHarness3.processElement(new StreamRecord<>("test4", 0L)); + checkFs(outDir, 4, 0, 0, 0); + + // intentionally we snapshot them in the reverse order so that the states are shuffled + OperatorStateHandles mergedSnapshot = AbstractStreamOperatorTestHarness.repackageState( + testHarness3.snapshot(0, 0), + testHarness1.snapshot(0, 0), + testHarness2.snapshot(0, 0) + ); + + testHarness1 = createRescalingTestSink(outDir, 2, 0, 100); + testHarness1.setup(); + testHarness1.initializeState(mergedSnapshot); + testHarness1.open(); + + checkFs(outDir, 1, 0, 3, 3); + + testHarness2 = createRescalingTestSink(outDir, 2, 1, 100); + testHarness2.setup(); + testHarness2.initializeState(mergedSnapshot); + testHarness2.open(); + + checkFs(outDir, 0, 0, 4, 4); + } + + @Test + public void testScalingUp() throws Exception { + final File outDir = tempFolder.newFolder(); + + OneInputStreamOperatorTestHarness<String, Object> testHarness1 = createRescalingTestSink(outDir, 2, 0, 100); + testHarness1.setup(); + testHarness1.open(); + + OneInputStreamOperatorTestHarness<String, Object> testHarness2 = createRescalingTestSink(outDir, 2, 0, 100); + testHarness2.setup(); + testHarness2.open(); + + testHarness1.processElement(new StreamRecord<>("test1", 1L)); + testHarness1.processElement(new StreamRecord<>("test2", 1L)); + + checkFs(outDir, 2, 0, 0, 0); + + testHarness2.processElement(new StreamRecord<>("test3", 1L)); + testHarness2.processElement(new StreamRecord<>("test4", 1L)); + testHarness2.processElement(new StreamRecord<>("test5", 1L)); + + checkFs(outDir, 5, 0, 0, 0); + + // intentionally we snapshot them in the reverse order so that the states are shuffled + OperatorStateHandles mergedSnapshot = AbstractStreamOperatorTestHarness.repackageState( + testHarness2.snapshot(0, 0), + testHarness1.snapshot(0, 0) + ); + + testHarness1 = createRescalingTestSink(outDir, 3, 0, 100); + testHarness1.setup(); + testHarness1.initializeState(mergedSnapshot); + testHarness1.open(); + + checkFs(outDir, 2, 0, 3, 3); + + testHarness2 = createRescalingTestSink(outDir, 3, 1, 100); + testHarness2.setup(); + testHarness2.initializeState(mergedSnapshot); + testHarness2.open(); + + checkFs(outDir, 0, 0, 5, 5); + + OneInputStreamOperatorTestHarness<String, Object> testHarness3 = createRescalingTestSink(outDir, 3, 2, 100); + testHarness3.setup(); + testHarness3.initializeState(mergedSnapshot); + testHarness3.open(); + + testHarness1.processElement(new StreamRecord<>("test6", 0)); + testHarness2.processElement(new StreamRecord<>("test6", 0)); + testHarness3.processElement(new StreamRecord<>("test6", 0)); + + checkFs(outDir, 3, 0, 5, 5); + + testHarness1.snapshot(1, 0); + testHarness2.snapshot(1, 0); + testHarness3.snapshot(1, 0); + + testHarness1.close(); + testHarness2.close(); + testHarness3.close(); + + checkFs(outDir, 0, 3, 5, 5); + } + + private void checkFs(File outDir, int inprogress, int pending, int completed, int valid) throws IOException { + int inProg = 0; + int pend = 0; + int compl = 0; + int val = 0; + + for (File file: FileUtils.listFiles(outDir, null, true)) { if (file.getAbsolutePath().endsWith("crc")) { continue; } - if (file.getPath().contains("pending")) { - numPending++; - } else if (file.getName().startsWith("part")) { - numComplete++; + String path = file.getPath(); + if (path.endsWith(".in-progress")) { --- End diff -- these strings should be moved into constants fields in this test, and should also be used when creating the sink.
--- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---