azagrebin commented on a change in pull request #6520: [FLINK-10097][DataStream
API] Additional tests for StreamingFileSink
URL: https://github.com/apache/flink/pull/6520#discussion_r228928607
##########
File path:
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/RollingPolicyTest.java
##########
@@ -50,233 +43,234 @@
@Test
public void testDefaultRollingPolicy() throws Exception {
final File outDir = TEMP_FOLDER.newFolder();
+ final Path path = new Path(outDir.toURI());
- final RollingPolicy<Tuple2<String, Integer>, String>
rollingPolicy = DefaultRollingPolicy
- .create()
- .withMaxPartSize(10L)
- .withInactivityInterval(4L)
- .withRolloverInterval(11L)
- .build();
-
- try (
-
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Object> testHarness
= TestUtils.createCustomRescalingTestSink(
- outDir,
- 1,
- 0,
- 1L,
- new
TestUtils.TupleToStringBucketer(),
- new SimpleStringEncoder<>(),
- rollingPolicy,
- new
DefaultBucketFactoryImpl<>())
- ) {
- testHarness.setup();
- testHarness.open();
-
- testHarness.setProcessingTime(0L);
+ final RollingPolicy<String, String> originalRollingPolicy =
+ DefaultRollingPolicy
+ .create()
+ .withMaxPartSize(10L)
+ .withInactivityInterval(4L)
+ .withRolloverInterval(11L)
+ .build();
- testHarness.processElement(new
StreamRecord<>(Tuple2.of("test1", 1), 1L));
- TestUtils.checkLocalFs(outDir, 1, 0);
+ final MethodCallCountingPolicyWrapper<String, String>
rollingPolicy =
+ new
MethodCallCountingPolicyWrapper<>(originalRollingPolicy);
- // roll due to size
- testHarness.processElement(new
StreamRecord<>(Tuple2.of("test1", 2), 2L));
- TestUtils.checkLocalFs(outDir, 1, 0);
+ final Buckets<String, String> buckets = createBuckets(path,
rollingPolicy);
- testHarness.processElement(new
StreamRecord<>(Tuple2.of("test1", 3), 3L));
- TestUtils.checkLocalFs(outDir, 2, 0);
+ rollingPolicy.verifyCallCounters(0L, 0L, 0L, 0L, 0L, 0L);
- // roll due to inactivity
- testHarness.setProcessingTime(7L);
+ // these two will fill up the first in-progress file and at the
third it will roll ...
+ buckets.onElement("test1", new TestUtils.MockSinkContext(1L,
1L, 1L));
+ buckets.onElement("test1", new TestUtils.MockSinkContext(2L,
1L, 2L));
+ rollingPolicy.verifyCallCounters(0L, 0L, 1L, 0L, 0L, 0L);
- testHarness.processElement(new
StreamRecord<>(Tuple2.of("test1", 4), 4L));
- TestUtils.checkLocalFs(outDir, 3, 0);
+ buckets.onElement("test1", new TestUtils.MockSinkContext(3L,
1L, 3L));
+ rollingPolicy.verifyCallCounters(0L, 0L, 2L, 1L, 0L, 0L);
- // roll due to rollover interval
- testHarness.setProcessingTime(20L);
+ // still no time to roll
+ buckets.onProcessingTime(5L);
+ rollingPolicy.verifyCallCounters(0L, 0L, 2L, 1L, 1L, 0L);
- testHarness.processElement(new
StreamRecord<>(Tuple2.of("test1", 5), 5L));
- TestUtils.checkLocalFs(outDir, 4, 0);
+ // roll due to inactivity
+ buckets.onProcessingTime(7L);
+ rollingPolicy.verifyCallCounters(0L, 0L, 2L, 1L, 2L, 1L);
- // we take a checkpoint but we should not roll.
- testHarness.snapshot(1L, 1L);
+ buckets.onElement("test1", new TestUtils.MockSinkContext(3L,
1L, 3L));
- TestUtils.checkLocalFs(outDir, 4, 0);
+ // roll due to rollover interval
+ buckets.onProcessingTime(20L);
+ rollingPolicy.verifyCallCounters(0L, 0L, 2L, 1L, 3L, 2L);
- // acknowledge the checkpoint, so publish the 3 closed
files, but not the open one.
- testHarness.notifyOfCompletedCheckpoint(1L);
- TestUtils.checkLocalFs(outDir, 1, 3);
- }
+ // we take a checkpoint but we should not roll.
+ buckets.snapshotState(1L, new TestUtils.MockListState<>(), new
TestUtils.MockListState<>());
+ rollingPolicy.verifyCallCounters(0L, 0L, 2L, 1L, 3L, 2L);
}
@Test
public void testRollOnCheckpointPolicy() throws Exception {
final File outDir = TEMP_FOLDER.newFolder();
+ final Path path = new Path(outDir.toURI());
- final RollingPolicy<Tuple2<String, Integer>, String>
rollingPolicy = OnCheckpointRollingPolicy.build();
-
- try (
-
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Object> testHarness
= TestUtils.createCustomRescalingTestSink(
- outDir,
- 1,
- 0,
- 10L,
- new
TestUtils.TupleToStringBucketer(),
- new SimpleStringEncoder<>(),
- rollingPolicy,
- new
DefaultBucketFactoryImpl<>())
- ) {
- testHarness.setup();
- testHarness.open();
+ final MethodCallCountingPolicyWrapper<String, String>
rollingPolicy =
+ new
MethodCallCountingPolicyWrapper<>(OnCheckpointRollingPolicy.build());
- testHarness.setProcessingTime(0L);
+ final Buckets<String, String> buckets = createBuckets(path,
rollingPolicy);
- testHarness.processElement(new
StreamRecord<>(Tuple2.of("test2", 1), 1L));
+ rollingPolicy.verifyCallCounters(0L, 0L, 0L, 0L, 0L, 0L);
- testHarness.processElement(new
StreamRecord<>(Tuple2.of("test1", 1), 1L));
- testHarness.processElement(new
StreamRecord<>(Tuple2.of("test1", 2), 2L));
- TestUtils.checkLocalFs(outDir, 2, 0);
+ // the following 2 elements will close a part file because of
size...
Review comment:
the comment is probably from the next test
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
