You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by kk...@apache.org on 2018/11/01 08:02:06 UTC
[flink] branch master updated: [FLINK-10097][DataStream API]
Additional tests for StreamingFileSink.
This is an automated email from the ASF dual-hosted git repository.
kkloudas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 9ad932c [FLINK-10097][DataStream API] Additional tests for StreamingFileSink.
9ad932c is described below
commit 9ad932c52056d0ed7e96e1c385403a529e6a8edb
Author: kkloudas <kk...@gmail.com>
AuthorDate: Sat Aug 4 22:24:58 2018 +0200
[FLINK-10097][DataStream API] Additional tests for StreamingFileSink.
---
.../api/functions/sink/filesystem/Bucket.java | 25 +-
.../api/functions/sink/filesystem/Buckets.java | 20 ++
.../sink/filesystem/StreamingFileSink.java | 3 -
.../rollingpolicies/DefaultRollingPolicy.java | 4 +-
.../api/functions/sink/filesystem/BucketsTest.java | 339 ++++++++++++++++--
.../filesystem/LocalStreamingFileSinkTest.java | 132 -------
.../sink/filesystem/RollingPolicyTest.java | 381 ++++++++++-----------
.../api/functions/sink/filesystem/TestUtils.java | 113 ++++++
8 files changed, 656 insertions(+), 361 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Bucket.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Bucket.java
index 6187e68..65a7628 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Bucket.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Bucket.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.api.functions.sink.filesystem;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.fs.RecoverableWriter;
import org.apache.flink.util.Preconditions;
@@ -26,6 +27,8 @@ import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
@@ -59,10 +62,11 @@ public class Bucket<IN, BucketID> {
private final RollingPolicy<IN, BucketID> rollingPolicy;
- private final Map<Long, List<RecoverableWriter.CommitRecoverable>> pendingPartsPerCheckpoint = new HashMap<>();
+ private final Map<Long, List<RecoverableWriter.CommitRecoverable>> pendingPartsPerCheckpoint;
private long partCounter;
+ @Nullable
private PartFileWriter<IN, BucketID> inProgressPart;
private List<RecoverableWriter.CommitRecoverable> pendingPartsForCurrentCheckpoint;
@@ -88,6 +92,7 @@ public class Bucket<IN, BucketID> {
this.rollingPolicy = Preconditions.checkNotNull(rollingPolicy);
this.pendingPartsForCurrentCheckpoint = new ArrayList<>();
+ this.pendingPartsPerCheckpoint = new HashMap<>();
}
/**
@@ -277,6 +282,24 @@ public class Bucket<IN, BucketID> {
}
}
+ // --------------------------- Testing Methods -----------------------------
+
+ @VisibleForTesting
+ Map<Long, List<RecoverableWriter.CommitRecoverable>> getPendingPartsPerCheckpoint() {
+ return pendingPartsPerCheckpoint;
+ }
+
+ @Nullable
+ @VisibleForTesting
+ PartFileWriter<IN, BucketID> getInProgressPart() {
+ return inProgressPart;
+ }
+
+ @VisibleForTesting
+ List<RecoverableWriter.CommitRecoverable> getPendingPartsForCurrentCheckpoint() {
+ return pendingPartsForCurrentCheckpoint;
+ }
+
// --------------------------- Static Factory Methods -----------------------------
/**
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Buckets.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Buckets.java
index 2aca841..d08bc2a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Buckets.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Buckets.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.api.functions.sink.filesystem;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
@@ -186,6 +187,10 @@ public class Buckets<IN, BucketID> {
}
private void updateActiveBucketId(final BucketID bucketId, final Bucket<IN, BucketID> restoredBucket) throws IOException {
+ if (!restoredBucket.isActive()) {
+ return;
+ }
+
final Bucket<IN, BucketID> bucket = activeBuckets.get(bucketId);
if (bucket != null) {
bucket.merge(restoredBucket);
@@ -224,6 +229,9 @@ public class Buckets<IN, BucketID> {
LOG.info("Subtask {} checkpointing for checkpoint with id={} (max part counter={}).",
subtaskIndex, checkpointId, maxPartCounter);
+ bucketStatesContainer.clear();
+ partCounterStateContainer.clear();
+
snapshotActiveBuckets(checkpointId, bucketStatesContainer);
partCounterStateContainer.add(maxPartCounter);
}
@@ -341,4 +349,16 @@ public class Buckets<IN, BucketID> {
return elementTimestamp;
}
}
+
+ // --------------------------- Testing Methods -----------------------------
+
+ @VisibleForTesting
+ public long getMaxPartCounter() {
+ return maxPartCounter;
+ }
+
+ @VisibleForTesting
+ Map<BucketID, Bucket<IN, BucketID>> getActiveBuckets() {
+ return activeBuckets;
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java
index 6f57fee..dc0b1c6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java
@@ -344,9 +344,6 @@ public class StreamingFileSink<IN>
public void snapshotState(FunctionSnapshotContext context) throws Exception {
Preconditions.checkState(bucketStates != null && maxPartCountersState != null, "sink has not been initialized");
- bucketStates.clear();
- maxPartCountersState.clear();
-
buckets.snapshotState(
context.getCheckpointId(),
bucketStates,
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/rollingpolicies/DefaultRollingPolicy.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/rollingpolicies/DefaultRollingPolicy.java
index 7c75f1c..f890326 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/rollingpolicies/DefaultRollingPolicy.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/rollingpolicies/DefaultRollingPolicy.java
@@ -78,8 +78,8 @@ public final class DefaultRollingPolicy<IN, BucketID> implements RollingPolicy<I
@Override
public boolean shouldRollOnProcessingTime(final PartFileInfo<BucketID> partFileState, final long currentTime) {
- return currentTime - partFileState.getCreationTime() > rolloverInterval ||
- currentTime - partFileState.getLastUpdateTime() > inactivityInterval;
+ return currentTime - partFileState.getCreationTime() >= rolloverInterval ||
+ currentTime - partFileState.getLastUpdateTime() >= inactivityInterval;
}
/**
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketsTest.java
index 25622d1..aee3621 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketsTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketsTest.java
@@ -19,17 +19,26 @@
package org.apache.flink.streaming.api.functions.sink.filesystem;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
+import org.apache.flink.api.common.state.ListState;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.io.SimpleVersionedSerializer;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.functions.sink.filesystem.TestUtils.MockListState;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
+import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
+import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
+import org.hamcrest.Description;
+import org.hamcrest.TypeSafeMatcher;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+
+import static org.hamcrest.MatcherAssert.assertThat;
/**
* Tests for {@link Buckets}.
@@ -40,46 +49,287 @@ public class BucketsTest {
public static final TemporaryFolder TEMP_FOLDER = new TemporaryFolder();
@Test
- public void testContextPassingNormalExecution() throws Exception {
- testCorrectPassingOfContext(1L, 2L, 3L);
- }
+ public void testSnapshotAndRestore() throws Exception {
+ final File outDir = TEMP_FOLDER.newFolder();
+ final Path path = new Path(outDir.toURI());
- @Test
- public void testContextPassingNullTimestamp() throws Exception {
- testCorrectPassingOfContext(null, 2L, 3L);
- }
+ final RollingPolicy<String, String> onCheckpointRollingPolicy = OnCheckpointRollingPolicy.build();
- private void testCorrectPassingOfContext(Long timestamp, long watermark, long processingTime) throws Exception {
- final File outDir = TEMP_FOLDER.newFolder();
+ final Buckets<String, String> buckets = createBuckets(path, onCheckpointRollingPolicy, 0);
+
+ final ListState<byte[]> bucketStateContainer = new MockListState<>();
+ final ListState<Long> partCounterContainer = new MockListState<>();
+
+ buckets.onElement("test1", new TestUtils.MockSinkContext(null, 1L, 2L));
+ buckets.snapshotState(0L, bucketStateContainer, partCounterContainer);
+
+ assertThat(buckets.getActiveBuckets().get("test1"), hasSinglePartFileToBeCommittedOnCheckpointAck(path, "test1"));
+
+ buckets.onElement("test2", new TestUtils.MockSinkContext(null, 1L, 2L));
+ buckets.snapshotState(1L, bucketStateContainer, partCounterContainer);
+
+ assertThat(buckets.getActiveBuckets().get("test1"), hasSinglePartFileToBeCommittedOnCheckpointAck(path, "test1"));
+ assertThat(buckets.getActiveBuckets().get("test2"), hasSinglePartFileToBeCommittedOnCheckpointAck(path, "test2"));
+
+ Buckets<String, String> restoredBuckets =
+ restoreBuckets(path, onCheckpointRollingPolicy, 0, bucketStateContainer, partCounterContainer);
- final Long expectedTimestamp = timestamp;
- final long expectedWatermark = watermark;
- final long expectedProcessingTime = processingTime;
+ final Map<String, Bucket<String, String>> activeBuckets = restoredBuckets.getActiveBuckets();
- final Buckets<String, String> buckets = StreamingFileSink
- .<String>forRowFormat(new Path(outDir.toURI()), new SimpleStringEncoder<>())
- .withBucketAssigner(new VarifyingBucketer(expectedTimestamp, expectedWatermark, expectedProcessingTime))
- .createBuckets(2);
+ // because we commit pending files for previous checkpoints upon recovery
+ Assert.assertTrue(activeBuckets.isEmpty());
+ }
- buckets.onElement("TEST", new SinkFunction.Context() {
+ private static TypeSafeMatcher<Bucket<String, String>> hasSinglePartFileToBeCommittedOnCheckpointAck(final Path testTmpPath, final String bucketId) {
+ return new TypeSafeMatcher<Bucket<String, String>>() {
@Override
- public long currentProcessingTime() {
- return expectedProcessingTime;
+ protected boolean matchesSafely(Bucket<String, String> bucket) {
+ return bucket.getBucketId().equals(bucketId) &&
+ bucket.getBucketPath().equals(new Path(testTmpPath, bucketId)) &&
+ bucket.getInProgressPart() == null &&
+ bucket.getPendingPartsForCurrentCheckpoint().isEmpty() &&
+ bucket.getPendingPartsPerCheckpoint().size() == 1;
}
@Override
- public long currentWatermark() {
- return expectedWatermark;
+ public void describeTo(Description description) {
+ description.appendText("a Bucket with a single pending part file @ ")
+ .appendValue(new Path(testTmpPath, bucketId))
+ .appendText("'");
}
+ };
+ }
- @Override
- public Long timestamp() {
- return expectedTimestamp;
+ @Test
+ public void testMergeAtScaleInAndMaxCounterAtRecovery() throws Exception {
+ final File outDir = TEMP_FOLDER.newFolder();
+ final Path path = new Path(outDir.toURI());
+
+ final RollingPolicy<String, String> onCheckpointRP =
+ DefaultRollingPolicy
+ .create()
+ .withMaxPartSize(7L) // roll with 2 elements
+ .build();
+
+ final MockListState<byte[]> bucketStateContainerOne = new MockListState<>();
+ final MockListState<byte[]> bucketStateContainerTwo = new MockListState<>();
+
+ final MockListState<Long> partCounterContainerOne = new MockListState<>();
+ final MockListState<Long> partCounterContainerTwo = new MockListState<>();
+
+ final Buckets<String, String> bucketsOne = createBuckets(path, onCheckpointRP, 0);
+ final Buckets<String, String> bucketsTwo = createBuckets(path, onCheckpointRP, 1);
+
+ bucketsOne.onElement("test1", new TestUtils.MockSinkContext(null, 1L, 2L));
+ bucketsOne.snapshotState(0L, bucketStateContainerOne, partCounterContainerOne);
+
+ Assert.assertEquals(1L, bucketsOne.getMaxPartCounter());
+
+ // make sure we have one in-progress file here
+ Assert.assertNotNull(bucketsOne.getActiveBuckets().get("test1").getInProgressPart());
+
+ // add a couple of in-progress files so that the part counter increases.
+ bucketsTwo.onElement("test1", new TestUtils.MockSinkContext(null, 1L, 2L));
+ bucketsTwo.onElement("test1", new TestUtils.MockSinkContext(null, 1L, 2L));
+
+ bucketsTwo.onElement("test1", new TestUtils.MockSinkContext(null, 1L, 2L));
+
+ bucketsTwo.snapshotState(0L, bucketStateContainerTwo, partCounterContainerTwo);
+
+ Assert.assertEquals(2L, bucketsTwo.getMaxPartCounter());
+
+ // make sure we have one in-progress file here and a pending
+ Assert.assertEquals(1L, bucketsTwo.getActiveBuckets().get("test1").getPendingPartsPerCheckpoint().size());
+ Assert.assertNotNull(bucketsTwo.getActiveBuckets().get("test1").getInProgressPart());
+
+ final ListState<byte[]> mergedBucketStateContainer = new MockListState<>();
+ final ListState<Long> mergedPartCounterContainer = new MockListState<>();
+
+ mergedBucketStateContainer.addAll(bucketStateContainerOne.getBackingList());
+ mergedBucketStateContainer.addAll(bucketStateContainerTwo.getBackingList());
+
+ mergedPartCounterContainer.addAll(partCounterContainerOne.getBackingList());
+ mergedPartCounterContainer.addAll(partCounterContainerTwo.getBackingList());
+
+ final Buckets<String, String> restoredBuckets =
+ restoreBuckets(path, onCheckpointRP, 0, mergedBucketStateContainer, mergedPartCounterContainer);
+
+ // we get the maximum of the previous tasks
+ Assert.assertEquals(2L, restoredBuckets.getMaxPartCounter());
+
+ final Map<String, Bucket<String, String>> activeBuckets = restoredBuckets.getActiveBuckets();
+ Assert.assertEquals(1L, activeBuckets.size());
+ Assert.assertTrue(activeBuckets.keySet().contains("test1"));
+
+ final Bucket<String, String> bucket = activeBuckets.get("test1");
+ Assert.assertEquals("test1", bucket.getBucketId());
+ Assert.assertEquals(new Path(path, "test1"), bucket.getBucketPath());
+
+ Assert.assertNotNull(bucket.getInProgressPart()); // the restored part file
+
+ // this is due to the Bucket#merge(). The in progress file of one
+ // of the previous tasks is put in the list of pending files.
+ Assert.assertEquals(1L, bucket.getPendingPartsForCurrentCheckpoint().size());
+
+ // we commit the pending for previous checkpoints
+ Assert.assertTrue(bucket.getPendingPartsPerCheckpoint().isEmpty());
+ }
+
+ @Test
+ public void testOnProcessingTime() throws Exception {
+ final File outDir = TEMP_FOLDER.newFolder();
+ final Path path = new Path(outDir.toURI());
+
+ final OnProcessingTimePolicy<String, String> rollOnProcessingTimeCountingPolicy =
+ new OnProcessingTimePolicy<>(2L);
+
+ final Buckets<String, String> buckets =
+ createBuckets(path, rollOnProcessingTimeCountingPolicy, 0);
+
+ // it takes the current processing time of the context for the creation time,
+ // and for the last modification time.
+ buckets.onElement("test", new TestUtils.MockSinkContext(1L, 2L , 3L));
+
+ // now it should roll
+ buckets.onProcessingTime(7L);
+ Assert.assertEquals(1L, rollOnProcessingTimeCountingPolicy.getOnProcessingTimeRollCounter());
+
+ final Map<String, Bucket<String, String>> activeBuckets = buckets.getActiveBuckets();
+ Assert.assertEquals(1L, activeBuckets.size());
+ Assert.assertTrue(activeBuckets.keySet().contains("test"));
+
+ final Bucket<String, String> bucket = activeBuckets.get("test");
+ Assert.assertEquals("test", bucket.getBucketId());
+ Assert.assertEquals(new Path(path, "test"), bucket.getBucketPath());
+ Assert.assertEquals("test", bucket.getBucketId());
+
+ Assert.assertNull(bucket.getInProgressPart());
+ Assert.assertEquals(1L, bucket.getPendingPartsForCurrentCheckpoint().size());
+ Assert.assertTrue(bucket.getPendingPartsPerCheckpoint().isEmpty());
+ }
+
+ @Test
+ public void testBucketIsRemovedWhenNotActive() throws Exception {
+ final File outDir = TEMP_FOLDER.newFolder();
+ final Path path = new Path(outDir.toURI());
+
+ final OnProcessingTimePolicy<String, String> rollOnProcessingTimeCountingPolicy =
+ new OnProcessingTimePolicy<>(2L);
+
+ final Buckets<String, String> buckets =
+ createBuckets(path, rollOnProcessingTimeCountingPolicy, 0);
+
+ // it takes the current processing time of the context for the creation time, and for the last modification time.
+ buckets.onElement("test", new TestUtils.MockSinkContext(1L, 2L , 3L));
+
+ // now it should roll
+ buckets.onProcessingTime(7L);
+ Assert.assertEquals(1L, rollOnProcessingTimeCountingPolicy.getOnProcessingTimeRollCounter());
+
+ buckets.snapshotState(0L, new MockListState<>(), new MockListState<>());
+ buckets.commitUpToCheckpoint(0L);
+
+ Assert.assertTrue(buckets.getActiveBuckets().isEmpty());
+ }
+
+ @Test
+ public void testPartCounterAfterBucketResurrection() throws Exception {
+ final File outDir = TEMP_FOLDER.newFolder();
+ final Path path = new Path(outDir.toURI());
+
+ final OnProcessingTimePolicy<String, String> rollOnProcessingTimeCountingPolicy =
+ new OnProcessingTimePolicy<>(2L);
+
+ final Buckets<String, String> buckets =
+ createBuckets(path, rollOnProcessingTimeCountingPolicy, 0);
+
+ // it takes the current processing time of the context for the creation time, and for the last modification time.
+ buckets.onElement("test", new TestUtils.MockSinkContext(1L, 2L , 3L));
+ Assert.assertEquals(1L, buckets.getActiveBuckets().get("test").getPartCounter());
+
+ // now it should roll
+ buckets.onProcessingTime(7L);
+ Assert.assertEquals(1L, rollOnProcessingTimeCountingPolicy.getOnProcessingTimeRollCounter());
+ Assert.assertEquals(1L, buckets.getActiveBuckets().get("test").getPartCounter());
+
+ buckets.snapshotState(0L, new MockListState<>(), new MockListState<>());
+ buckets.commitUpToCheckpoint(0L);
+
+ Assert.assertTrue(buckets.getActiveBuckets().isEmpty());
+
+ buckets.onElement("test", new TestUtils.MockSinkContext(2L, 3L , 4L));
+ Assert.assertEquals(2L, buckets.getActiveBuckets().get("test").getPartCounter());
+ }
+
+ private static class OnProcessingTimePolicy<IN, BucketID> implements RollingPolicy<IN, BucketID> {
+
+ private static final long serialVersionUID = 1L;
+
+ private int onProcessingTimeRollCounter = 0;
+
+ private final long rolloverInterval;
+
+ OnProcessingTimePolicy(final long rolloverInterval) {
+ this.rolloverInterval = rolloverInterval;
+ }
+
+ public int getOnProcessingTimeRollCounter() {
+ return onProcessingTimeRollCounter;
+ }
+
+ @Override
+ public boolean shouldRollOnCheckpoint(PartFileInfo<BucketID> partFileState) {
+ return false;
+ }
+
+ @Override
+ public boolean shouldRollOnEvent(PartFileInfo<BucketID> partFileState, IN element) {
+ return false;
+ }
+
+ @Override
+ public boolean shouldRollOnProcessingTime(PartFileInfo<BucketID> partFileState, long currentTime) {
+ boolean result = currentTime - partFileState.getCreationTime() >= rolloverInterval;
+ if (result) {
+ onProcessingTimeRollCounter++;
}
- });
+ return result;
+ }
}
- private static class VarifyingBucketer implements BucketAssigner<String, String> {
+ @Test
+ public void testContextPassingNormalExecution() throws Exception {
+ testCorrectTimestampPassingInContext(1L, 2L, 3L);
+ }
+
+ @Test
+ public void testContextPassingNullTimestamp() throws Exception {
+ testCorrectTimestampPassingInContext(null, 2L, 3L);
+ }
+
+ private void testCorrectTimestampPassingInContext(Long timestamp, long watermark, long processingTime) throws Exception {
+ final File outDir = TEMP_FOLDER.newFolder();
+ final Path path = new Path(outDir.toURI());
+
+ final Buckets<String, String> buckets = new Buckets<>(
+ path,
+ new VerifyingBucketAssigner(timestamp, watermark, processingTime),
+ new DefaultBucketFactoryImpl<>(),
+ new RowWisePartWriter.Factory<>(new SimpleStringEncoder<>()),
+ DefaultRollingPolicy.create().build(),
+ 2
+ );
+
+ buckets.onElement(
+ "test",
+ new TestUtils.MockSinkContext(
+ timestamp,
+ watermark,
+ processingTime)
+ );
+ }
+
+ private static class VerifyingBucketAssigner implements BucketAssigner<String, String> {
private static final long serialVersionUID = 7729086510972377578L;
@@ -87,7 +337,7 @@ public class BucketsTest {
private final long expectedWatermark;
private final long expectedProcessingTime;
- VarifyingBucketer(
+ VerifyingBucketAssigner(
final Long expectedTimestamp,
final long expectedWatermark,
final long expectedProcessingTime
@@ -98,7 +348,7 @@ public class BucketsTest {
}
@Override
- public String getBucketId(String element, Context context) {
+ public String getBucketId(String element, BucketAssigner.Context context) {
final Long elementTimestamp = context.timestamp();
final long watermark = context.currentWatermark();
final long processingTime = context.currentProcessingTime();
@@ -115,4 +365,35 @@ public class BucketsTest {
return SimpleVersionedStringSerializer.INSTANCE;
}
}
+
+ // ------------------------------- Utility Methods --------------------------------
+
+ private static Buckets<String, String> createBuckets(
+ final Path basePath,
+ final RollingPolicy<String, String> rollingPolicy,
+ final int subtaskIdx
+ ) throws IOException {
+
+ return new Buckets<>(
+ basePath,
+ new TestUtils.StringIdentityBucketAssigner(),
+ new DefaultBucketFactoryImpl<>(),
+ new RowWisePartWriter.Factory<>(new SimpleStringEncoder<>()),
+ rollingPolicy,
+ subtaskIdx
+ );
+ }
+
+ private static Buckets<String, String> restoreBuckets(
+ final Path basePath,
+ final RollingPolicy<String, String> rollingPolicy,
+ final int subtaskIdx,
+ final ListState<byte[]> bucketState,
+ final ListState<Long> partCounterState
+ ) throws Exception {
+
+ final Buckets<String, String> restoredBuckets = createBuckets(basePath, rollingPolicy, subtaskIdx);
+ restoredBuckets.initializeState(bucketState, partCounterState);
+ return restoredBuckets;
+ }
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/LocalStreamingFileSinkTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/LocalStreamingFileSinkTest.java
index 8bb35ff..9c3f946 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/LocalStreamingFileSinkTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/LocalStreamingFileSinkTest.java
@@ -18,12 +18,8 @@
package org.apache.flink.streaming.api.functions.sink.filesystem;
-import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.core.fs.Path;
-import org.apache.flink.core.fs.RecoverableWriter;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
-import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
@@ -35,7 +31,6 @@ import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import java.io.File;
-import java.io.IOException;
import java.util.Map;
/**
@@ -494,131 +489,4 @@ public class LocalStreamingFileSinkTest extends TestLogger {
Assert.assertEquals(3L, counter);
}
}
-
- @Test
- public void testMaxCounterUponRecovery() throws Exception {
- final File outDir = TEMP_FOLDER.newFolder();
-
- OperatorSubtaskState mergedSnapshot;
-
- final TestBucketFactoryImpl first = new TestBucketFactoryImpl();
- final TestBucketFactoryImpl second = new TestBucketFactoryImpl();
-
- final RollingPolicy<Tuple2<String, Integer>, String> rollingPolicy = DefaultRollingPolicy
- .create()
- .withMaxPartSize(2L)
- .withRolloverInterval(100L)
- .withInactivityInterval(100L)
- .build();
-
- try (
- OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Object> testHarness1 = TestUtils.createCustomRescalingTestSink(
- outDir, 2, 0, 10L, new TestUtils.TupleToStringBucketer(), new SimpleStringEncoder<>(), rollingPolicy, first);
- OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Object> testHarness2 = TestUtils.createCustomRescalingTestSink(
- outDir, 2, 1, 10L, new TestUtils.TupleToStringBucketer(), new SimpleStringEncoder<>(), rollingPolicy, second)
- ) {
- testHarness1.setup();
- testHarness1.open();
-
- testHarness2.setup();
- testHarness2.open();
-
- // we only put elements in one task.
- testHarness1.processElement(new StreamRecord<>(Tuple2.of("test1", 0), 0L));
- testHarness1.processElement(new StreamRecord<>(Tuple2.of("test1", 0), 0L));
- testHarness1.processElement(new StreamRecord<>(Tuple2.of("test1", 0), 0L));
- TestUtils.checkLocalFs(outDir, 3, 0);
-
- // intentionally we snapshot them in the reverse order so that the states are shuffled
- mergedSnapshot = AbstractStreamOperatorTestHarness.repackageState(
- testHarness2.snapshot(0L, 0L),
- testHarness1.snapshot(0L, 0L)
- );
- }
-
- final TestBucketFactoryImpl firstRecovered = new TestBucketFactoryImpl();
- final TestBucketFactoryImpl secondRecovered = new TestBucketFactoryImpl();
-
- try (
- OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Object> testHarness1 = TestUtils.createCustomRescalingTestSink(
- outDir, 2, 0, 10L, new TestUtils.TupleToStringBucketer(), new SimpleStringEncoder<>(), rollingPolicy, firstRecovered);
- OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Object> testHarness2 = TestUtils.createCustomRescalingTestSink(
- outDir, 2, 1, 10L, new TestUtils.TupleToStringBucketer(), new SimpleStringEncoder<>(), rollingPolicy, secondRecovered)
- ) {
- testHarness1.setup();
- testHarness1.initializeState(mergedSnapshot);
- testHarness1.open();
-
- // we have to send an element so that the factory updates its counter.
- testHarness1.processElement(new StreamRecord<>(Tuple2.of("test4", 0), 0L));
-
- Assert.assertEquals(3L, firstRecovered.getInitialCounter());
- TestUtils.checkLocalFs(outDir, 1, 3);
-
- testHarness2.setup();
- testHarness2.initializeState(mergedSnapshot);
- testHarness2.open();
-
- // we have to send an element so that the factory updates its counter.
- testHarness2.processElement(new StreamRecord<>(Tuple2.of("test2", 0), 0L));
-
- Assert.assertEquals(3L, secondRecovered.getInitialCounter());
- TestUtils.checkLocalFs(outDir, 2, 3);
- }
- }
-
- ////////////////////// Helper Methods //////////////////////
-
- static class TestBucketFactoryImpl extends DefaultBucketFactoryImpl<Tuple2<String, Integer>, String> {
-
- private static final long serialVersionUID = 2794824980604027930L;
-
- private long initialCounter = -1L;
-
- @Override
- public Bucket<Tuple2<String, Integer>, String> getNewBucket(
- final RecoverableWriter fsWriter,
- final int subtaskIndex,
- final String bucketId,
- final Path bucketPath,
- final long initialPartCounter,
- final PartFileWriter.PartFileFactory<Tuple2<String, Integer>, String> partFileWriterFactory,
- final RollingPolicy<Tuple2<String, Integer>, String> rollingPolicy) {
-
- this.initialCounter = initialPartCounter;
-
- return super.getNewBucket(
- fsWriter,
- subtaskIndex,
- bucketId,
- bucketPath,
- initialPartCounter,
- partFileWriterFactory,
- rollingPolicy);
- }
-
- @Override
- public Bucket<Tuple2<String, Integer>, String> restoreBucket(
- final RecoverableWriter fsWriter,
- final int subtaskIndex,
- final long initialPartCounter,
- final PartFileWriter.PartFileFactory<Tuple2<String, Integer>, String> partFileWriterFactory,
- final RollingPolicy<Tuple2<String, Integer>, String> rollingPolicy,
- final BucketState<String> bucketState) throws IOException {
-
- this.initialCounter = initialPartCounter;
-
- return super.restoreBucket(
- fsWriter,
- subtaskIndex,
- initialPartCounter,
- partFileWriterFactory,
- rollingPolicy,
- bucketState);
- }
-
- public long getInitialCounter() {
- return initialCounter;
- }
- }
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/RollingPolicyTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/RollingPolicyTest.java
index 851b682..1f4c7e5 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/RollingPolicyTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/RollingPolicyTest.java
@@ -19,13 +19,11 @@
package org.apache.flink.streaming.api.functions.sink.filesystem;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.util.Preconditions;
-import org.apache.commons.io.FileUtils;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
@@ -33,11 +31,6 @@ import org.junit.rules.TemporaryFolder;
import java.io.File;
import java.io.IOException;
-import java.util.Objects;
-
-import static org.hamcrest.CoreMatchers.containsString;
-import static org.hamcrest.CoreMatchers.either;
-import static org.hamcrest.CoreMatchers.equalTo;
/**
* Tests for different {@link RollingPolicy rolling policies}.
@@ -50,233 +43,233 @@ public class RollingPolicyTest {
@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);
+ buckets.onElement("test1", new TestUtils.MockSinkContext(1L, 1L, 2L));
+ buckets.onElement("test1", new TestUtils.MockSinkContext(2L, 1L, 2L));
+ buckets.onElement("test1", new TestUtils.MockSinkContext(3L, 1L, 3L));
- testHarness.processElement(new StreamRecord<>(Tuple2.of("test1", 3), 3L));
- TestUtils.checkLocalFs(outDir, 2, 0);
+ // ... we have a checkpoint so we roll ...
+ buckets.snapshotState(1L, new TestUtils.MockListState<>(), new TestUtils.MockListState<>());
+ rollingPolicy.verifyCallCounters(1L, 1L, 2L, 0L, 0L, 0L);
- // we take a checkpoint so we roll.
- testHarness.snapshot(1L, 1L);
+ // ... create a new in-progress file (before we had closed the last one so it was null)...
+ buckets.onElement("test1", new TestUtils.MockSinkContext(5L, 1L, 5L));
- for (File file: FileUtils.listFiles(outDir, null, true)) {
- if (Objects.equals(file.getParentFile().getName(), "test1")) {
- Assert.assertTrue(file.getName().contains(".part-0-1.inprogress."));
- } else if (Objects.equals(file.getParentFile().getName(), "test2")) {
- Assert.assertTrue(file.getName().contains(".part-0-0.inprogress."));
- }
- }
+ // ... we have a checkpoint so we roll ...
+ buckets.snapshotState(2L, new TestUtils.MockListState<>(), new TestUtils.MockListState<>());
+ rollingPolicy.verifyCallCounters(2L, 2L, 2L, 0L, 0L, 0L);
- // this will create a new part file
- testHarness.processElement(new StreamRecord<>(Tuple2.of("test1", 4), 4L));
- TestUtils.checkLocalFs(outDir, 3, 0);
-
- testHarness.notifyOfCompletedCheckpoint(1L);
- for (File file: FileUtils.listFiles(outDir, null, true)) {
- if (Objects.equals(file.getParentFile().getName(), "test1")) {
- Assert.assertTrue(
- file.getName().contains(".part-0-2.inprogress.") || file.getName().equals("part-0-1")
- );
- } else if (Objects.equals(file.getParentFile().getName(), "test2")) {
- Assert.assertEquals("part-0-0", file.getName());
- }
- }
-
- // and open and fill .part-0-2.inprogress
- testHarness.processElement(new StreamRecord<>(Tuple2.of("test1", 5), 5L));
- testHarness.processElement(new StreamRecord<>(Tuple2.of("test1", 6), 6L));
- TestUtils.checkLocalFs(outDir, 1, 2);
-
- // we take a checkpoint so we roll.
- testHarness.snapshot(2L, 2L);
-
- testHarness.processElement(new StreamRecord<>(Tuple2.of("test2", 7), 7L));
- TestUtils.checkLocalFs(outDir, 2, 2);
-
- for (File file: FileUtils.listFiles(outDir, null, true)) {
- if (Objects.equals(file.getParentFile().getName(), "test1")) {
- Assert.assertThat(
- file.getName(),
- either(containsString(".part-0-2.inprogress."))
- .or(equalTo("part-0-1"))
- );
- } else if (Objects.equals(file.getParentFile().getName(), "test2")) {
- Assert.assertThat(
- file.getName(),
- either(containsString(".part-0-3.inprogress."))
- .or(equalTo("part-0-0"))
- );
- }
- }
-
- // we acknowledge the last checkpoint so we should publish all but the latest in-progress file
- testHarness.notifyOfCompletedCheckpoint(2L);
-
- TestUtils.checkLocalFs(outDir, 1, 3);
- for (File file: FileUtils.listFiles(outDir, null, true)) {
- if (Objects.equals(file.getParentFile().getName(), "test1")) {
- Assert.assertThat(
- file.getName(),
- either(equalTo("part-0-2")).or(equalTo("part-0-1"))
- );
- } else if (Objects.equals(file.getParentFile().getName(), "test2")) {
- Assert.assertThat(
- file.getName(),
- either(containsString(".part-0-3.inprogress."))
- .or(equalTo("part-0-0"))
- );
- }
- }
- }
+ buckets.close();
}
@Test
public void testCustomRollingPolicy() throws Exception {
final File outDir = TEMP_FOLDER.newFolder();
+ final Path path = new Path(outDir.toURI());
- final RollingPolicy<Tuple2<String, Integer>, String> rollingPolicy = new RollingPolicy<Tuple2<String, Integer>, String>() {
+ final MethodCallCountingPolicyWrapper<String, String> rollingPolicy = new MethodCallCountingPolicyWrapper<>(
+ new RollingPolicy<String, String>() {
- private static final long serialVersionUID = 1L;
+ private static final long serialVersionUID = 1L;
- @Override
- public boolean shouldRollOnCheckpoint(PartFileInfo<String> partFileState) {
- return true;
- }
+ @Override
+ public boolean shouldRollOnCheckpoint(PartFileInfo<String> partFileState) {
+ return true;
+ }
- @Override
- public boolean shouldRollOnEvent(PartFileInfo<String> partFileState, Tuple2<String, Integer> element) throws IOException {
- // this means that 2 elements will close the part file.
- return partFileState.getSize() > 12L;
- }
+ @Override
+ public boolean shouldRollOnEvent(PartFileInfo<String> partFileState, String element) throws IOException {
+ // this means that 2 elements will close the part file.
+ return partFileState.getSize() > 9L;
+ }
- @Override
- public boolean shouldRollOnProcessingTime(PartFileInfo<String> partFileState, long currentTime) {
- return false;
- }
- };
-
- 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();
+ @Override
+ public boolean shouldRollOnProcessingTime(PartFileInfo<String> partFileState, long currentTime) {
+ return currentTime - partFileState.getLastUpdateTime() >= 10L;
+ }
+ });
+
+ final Buckets<String, String> buckets = createBuckets(path, rollingPolicy);
- testHarness.setProcessingTime(0L);
+ rollingPolicy.verifyCallCounters(0L, 0L, 0L, 0L, 0L, 0L);
- testHarness.processElement(new StreamRecord<>(Tuple2.of("test2", 1), 1L));
+ // the following 2 elements will close a part file because of size...
+ buckets.onElement("test1", new TestUtils.MockSinkContext(1L, 1L, 2L));
+ buckets.onElement("test1", new TestUtils.MockSinkContext(2L, 1L, 2L));
- // the following 2 elements will close a part file ...
- testHarness.processElement(new StreamRecord<>(Tuple2.of("test1", 1), 1L));
- testHarness.processElement(new StreamRecord<>(Tuple2.of("test1", 2), 2L));
+ // only one call because we have no open part file in the other incoming elements, so currentPartFile == null so we roll without checking the policy.
+ rollingPolicy.verifyCallCounters(0L, 0L, 1L, 0L, 0L, 0L);
- // ... and this one will open a new ...
- testHarness.processElement(new StreamRecord<>(Tuple2.of("test1", 3), 2L));
- TestUtils.checkLocalFs(outDir, 3, 0);
+ // ... and this one will trigger the roll and open a new part file...
+ buckets.onElement("test1", new TestUtils.MockSinkContext(2L, 1L, 2L));
+ rollingPolicy.verifyCallCounters(0L, 0L, 2L, 1L, 0L, 0L);
- // ... and all open part files should close here.
- testHarness.snapshot(1L, 1L);
+ // ... we have a checkpoint so we roll ...
+ buckets.snapshotState(1L, new TestUtils.MockListState<>(), new TestUtils.MockListState<>());
+ rollingPolicy.verifyCallCounters(1L, 1L, 2L, 1L, 0L, 0L);
- // this will create and fill out a new part file
- testHarness.processElement(new StreamRecord<>(Tuple2.of("test1", 4), 4L));
- testHarness.processElement(new StreamRecord<>(Tuple2.of("test1", 5), 5L));
- TestUtils.checkLocalFs(outDir, 4, 0);
+ // ... create a new in-progress file (before we had closed the last one so it was null)...
+ buckets.onElement("test1", new TestUtils.MockSinkContext(2L, 1L, 5L));
- // we take a checkpoint so we roll.
- testHarness.snapshot(2L, 2L);
+ // ... last modification time is 5L, so now we DON'T roll but we check ...
+ buckets.onProcessingTime(12L);
+ rollingPolicy.verifyCallCounters(1L, 1L, 2L, 1L, 1L, 0L);
+
+ // ... last modification time is 5L, so now we roll
+ buckets.onProcessingTime(16L);
+ rollingPolicy.verifyCallCounters(1L, 1L, 2L, 1L, 2L, 1L);
+
+ buckets.close();
+ }
+
+ // ------------------------------- Utility Methods --------------------------------
+
+ private static Buckets<String, String> createBuckets(
+ final Path basePath,
+ final MethodCallCountingPolicyWrapper<String, String> rollingPolicyToTest
+ ) throws IOException {
+
+ return new Buckets<>(
+ basePath,
+ new TestUtils.StringIdentityBucketAssigner(),
+ new DefaultBucketFactoryImpl<>(),
+ new RowWisePartWriter.Factory<>(new SimpleStringEncoder<>()),
+ rollingPolicyToTest,
+ 0
+ );
+ }
- // we acknowledge the first checkpoint so we should publish all but the latest in-progress file
- testHarness.notifyOfCompletedCheckpoint(1L);
- TestUtils.checkLocalFs(outDir, 1, 3);
+ /**
+ * A wrapper of a {@link RollingPolicy} which counts how many times each method of the policy was called
+ * and in how many of them it decided to roll.
+ */
+ private static class MethodCallCountingPolicyWrapper<IN, BucketID> implements RollingPolicy<IN, BucketID> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final RollingPolicy<IN, BucketID> originalPolicy;
+
+ private long onCheckpointCallCounter;
+ private long onCheckpointRollCounter;
+
+ private long onEventCallCounter;
+ private long onEventRollCounter;
+
+ private long onProcessingTimeCallCounter;
+ private long onProcessingTimeRollCounter;
+
+ MethodCallCountingPolicyWrapper(final RollingPolicy<IN, BucketID> policy) {
+ this.originalPolicy = Preconditions.checkNotNull(policy);
+
+ this.onCheckpointCallCounter = 0L;
+ this.onCheckpointRollCounter = 0L;
+
+ this.onEventCallCounter = 0L;
+ this.onEventRollCounter = 0L;
+
+ this.onProcessingTimeCallCounter = 0L;
+ this.onProcessingTimeRollCounter = 0L;
+ }
+
+ @Override
+ public boolean shouldRollOnCheckpoint(PartFileInfo<BucketID> partFileState) throws IOException {
+ final boolean shouldRoll = originalPolicy.shouldRollOnCheckpoint(partFileState);
+ this.onCheckpointCallCounter++;
+ if (shouldRoll) {
+ this.onCheckpointRollCounter++;
+ }
+ return shouldRoll;
+ }
+
+ @Override
+ public boolean shouldRollOnEvent(PartFileInfo<BucketID> partFileState, IN element) throws IOException {
+ final boolean shouldRoll = originalPolicy.shouldRollOnEvent(partFileState, element);
+ this.onEventCallCounter++;
+ if (shouldRoll) {
+ this.onEventRollCounter++;
+ }
+ return shouldRoll;
+ }
+
+ @Override
+ public boolean shouldRollOnProcessingTime(PartFileInfo<BucketID> partFileState, long currentTime) throws IOException {
+ final boolean shouldRoll = originalPolicy.shouldRollOnProcessingTime(partFileState, currentTime);
+ this.onProcessingTimeCallCounter++;
+ if (shouldRoll) {
+ this.onProcessingTimeRollCounter++;
+ }
+ return shouldRoll;
+ }
+
+ void verifyCallCounters(
+ final long onCheckpointCalls,
+ final long onCheckpointRolls,
+ final long onEventCalls,
+ final long onEventRolls,
+ final long onProcessingTimeCalls,
+ final long onProcessingTimeRolls
+ ) {
+ Assert.assertEquals(onCheckpointCalls, onCheckpointCallCounter);
+ Assert.assertEquals(onCheckpointRolls, onCheckpointRollCounter);
+ Assert.assertEquals(onEventCalls, onEventCallCounter);
+ Assert.assertEquals(onEventRolls, onEventRollCounter);
+ Assert.assertEquals(onProcessingTimeCalls, onProcessingTimeCallCounter);
+ Assert.assertEquals(onProcessingTimeRolls, onProcessingTimeRollCounter);
}
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/TestUtils.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/TestUtils.java
index bfbc120..9e33064 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/TestUtils.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/TestUtils.java
@@ -20,9 +20,11 @@ package org.apache.flink.streaming.api.functions.sink.filesystem;
import org.apache.flink.api.common.serialization.BulkWriter;
import org.apache.flink.api.common.serialization.Encoder;
+import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.api.operators.StreamSink;
@@ -31,11 +33,17 @@ import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.commons.io.FileUtils;
import org.junit.Assert;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
import java.util.Map;
/**
@@ -160,4 +168,109 @@ public class TestUtils {
return SimpleVersionedStringSerializer.INSTANCE;
}
}
+
+ /**
+ * A simple {@link BucketAssigner} that accepts {@code String}'s
+ * and returns the element itself as the bucket id.
+ */
+ static class StringIdentityBucketAssigner implements BucketAssigner<String, String> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public String getBucketId(String element, BucketAssigner.Context context) {
+ return element;
+ }
+
+ @Override
+ public SimpleVersionedSerializer<String> getSerializer() {
+ return SimpleVersionedStringSerializer.INSTANCE;
+ }
+ }
+
+ /**
+ * A mock {@link SinkFunction.Context} to be used in the tests.
+ */
+ static class MockSinkContext implements SinkFunction.Context {
+
+ @Nullable
+ private Long elementTimestamp;
+
+ private long watermark;
+
+ private long processingTime;
+
+ MockSinkContext(
+ @Nullable Long elementTimestamp,
+ long watermark,
+ long processingTime) {
+ this.elementTimestamp = elementTimestamp;
+ this.watermark = watermark;
+ this.processingTime = processingTime;
+ }
+
+ @Override
+ public long currentProcessingTime() {
+ return processingTime;
+ }
+
+ @Override
+ public long currentWatermark() {
+ return watermark;
+ }
+
+ @Nullable
+ @Override
+ public Long timestamp() {
+ return elementTimestamp;
+ }
+ }
+
+ /**
+ * A mock {@link ListState} used for testing the snapshot/restore cycle of the sink.
+ */
+ static class MockListState<T> implements ListState<T> {
+
+ private final List<T> backingList;
+
+ MockListState() {
+ this.backingList = new ArrayList<>();
+ }
+
+ public List<T> getBackingList() {
+ return backingList;
+ }
+
+ @Override
+ public void update(List<T> values) {
+ backingList.clear();
+ addAll(values);
+ }
+
+ @Override
+ public void addAll(List<T> values) {
+ backingList.addAll(values);
+ }
+
+ @Override
+ public Iterable<T> get() {
+ return new Iterable<T>() {
+
+ @Nonnull
+ @Override
+ public Iterator<T> iterator() {
+ return backingList.iterator();
+ }
+ };
+ }
+
+ @Override
+ public void add(T value) {
+ backingList.add(value);
+ }
+
+ @Override
+ public void clear() {
+ backingList.clear();
+ }
+ }
}