You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2020/11/30 15:15:47 UTC
[flink] branch release-1.12 updated (be972ef -> 56bfa33)
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a change to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git.
from be972ef [FLINK-20391] Set FORWARD_EDGES_PIPELINED for BATCH ExecutionMode
new 99a8891 [hotfix][fs-connector] Remove unused state serializer from FileWriter
new 4bd4e76 [hotfix][fs-connector] Fix the typo in the FileSinkITCase
new 836d422 [FLINK-20337] Let StatefulSinkWriterOperator load StreamingFileSink's state
new 073c952 [FLINK-20337] Extend StatefulSinkWriterOperator Javadoc
new d0757e9 [FLINK-20337] Let FileSink restore state from StreamingFileSink
new 684bdce [refactor] Factor test data generator/resolver out of BucketStateSerializerTest
new 63f7f3b [FLINK-20337] Add migration test for deserializing existing StreamingFileSink state with FileSink serializer
new 0528a21 [refactor] Factor common testing code out of FileSinkITBase
new c54f5a7 [FLINK-20337] Add ITCase for migrating from StreamingFileSink to FileSink
new 56bfa33 [refactor] Move FileSink migration tests to flink-connector-files
The 10 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
flink-connectors/flink-connector-files/pom.xml | 30 +++
.../apache/flink/connector/file/sink/FileSink.java | 14 +-
.../connector/file/sink/writer/FileWriter.java | 9 +-
.../file/sink/writer/FileWriterBucket.java | 12 +
.../file/sink/writer/FileWriterBucketState.java | 33 +++
.../writer/FileWriterBucketStateSerializer.java | 110 +++++++-
.../flink/connector/file/sink/FileSinkITBase.java | 93 +------
.../sink/StreamingExecutionFileSinkITCase.java | 2 +-
.../sink/utils/IntegerFileSinkTestDataUtils.java | 140 ++++++++++
.../file/sink/writer/FileSinkMigrationITCase.java | 287 +++++++++++++++++++++
...leWriterBucketStateSerializerMigrationTest.java | 285 ++++++++++++++++++++
.../FileWriterBucketStateSerializerTest.java | 4 +-
.../file/sink/writer/FileWriterBucketTest.java | 59 +++++
.../bucket-state-migration-test/empty-v1/snapshot | Bin
.../bucket-state-migration-test/empty-v2/snapshot | Bin
...inprogress.a88d5993-77bc-44ce-880b-9f2a43b59ab4 | 0
...inprogress.7c0f2bd7-3078-48e8-9af2-d8773fb949c5 | 0
...inprogress.6729a640-0585-4785-a652-89802950c663 | 0
...inprogress.b4bcb0e9-5c9e-45dd-8963-1b163343544d | 0
...inprogress.e1e9e48d-0db6-4dd7-8a4d-fb4ebe7ed8ac | 0
.../full-no-in-progress-v1-template/snapshot | Bin
...inprogress.1e22e72d-0ab2-493b-8b00-9edac4252cec | 0
...inprogress.3821f491-9fa1-48b2-b66b-655352a3c8ec | 0
...inprogress.0af18f41-d8f8-4a4e-a92e-de12851be20b | 0
...inprogress.a3d0f4d2-d6ad-4f83-ba62-ed4b1fa86db2 | 0
...inprogress.666acf3e-935c-4621-8171-f7c897496524 | 0
.../full-no-in-progress-v2-template/snapshot | Bin
...inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82 | 0
...inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c | 0
...inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97 | 0
...inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9 | 0
...inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7 | 0
...inprogress.32f5a28f-20e1-48da-9951-10e795133d64 | 0
.../full-v1-template/snapshot | Bin
...inprogress.9731063e-2b28-4701-8cc1-e706480b8022 | 0
...inprogress.1d423406-097a-4deb-bfde-d023d3477cd5 | 0
...inprogress.6a837aa3-4736-4098-a878-fdeffe227628 | 0
...inprogress.f121b73d-ac74-4fbd-b70d-f13e51c9132c | 0
...inprogress.a156884a-f090-4c3f-a271-0b63ab539c45 | 0
...inprogress.83c527c5-14dc-4d49-9f99-c915f2224f6a | 0
.../full-v2-template/snapshot | Bin
...inprogress.a70190d6-d080-43a8-b414-746b09d3a8a0 | 0
.../only-in-progress-v1/snapshot | Bin
...inprogress.10833090-dd8c-4e36-884d-bb9758a3a8ef | 0
.../only-in-progress-v2/snapshot | Bin
.../OutputStreamBasedPartFileWriter.java | 23 +-
.../operators/sink/StatefulSinkWriterOperator.java | 58 ++++-
.../sink/StatefulSinkWriterOperatorFactory.java | 18 +-
.../translators/SinkTransformationTranslator.java | 15 +-
.../sink/filesystem/BucketStateGenerator.java | 173 +++++++++++++
...Committer.java => BucketStatePathResolver.java} | 38 +--
.../sink/filesystem/BucketStateSerializerTest.java | 153 ++---------
.../sink/StatefulSinkWriterOperatorTest.java | 109 ++++++++
53 files changed, 1390 insertions(+), 275 deletions(-)
create mode 100644 flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/utils/IntegerFileSinkTestDataUtils.java
create mode 100644 flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileSinkMigrationITCase.java
create mode 100644 flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerMigrationTest.java
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/empty-v1/snapshot (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/empty-v2/snapshot (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-0.inprogress.a88d5993-77bc-44ce-880b-9f2a43b59ab4 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-1.inprogress.7c0f2bd7-3078-48e8-9af2-d8773fb949c5 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-2.inprogress.6729a640-0585-4785-a652-89802950c663 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-3.inprogress.b4bcb0e9-5c9e-45dd-8963-1b163343544d (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-4.inprogress.e1e9e48d-0db6-4dd7-8a4d-fb4ebe7ed8ac (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/snapshot (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-0.inprogress.1e22e72d-0ab2-493b-8b00-9edac4252cec (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-1.inprogress.3821f491-9fa1-48b2-b66b-655352a3c8ec (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-2.inprogress.0af18f41-d8f8-4a4e-a92e-de12851be20b (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-3.inprogress.a3d0f4d2-d6ad-4f83-ba62-ed4b1fa86db2 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-4.inprogress.666acf3e-935c-4621-8171-f7c897496524 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/snapshot (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-0.inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-1.inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-2.inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-3.inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-4.inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-5.inprogress.32f5a28f-20e1-48da-9951-10e795133d64 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v1-template/snapshot (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-0.inprogress.9731063e-2b28-4701-8cc1-e706480b8022 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-1.inprogress.1d423406-097a-4deb-bfde-d023d3477cd5 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-2.inprogress.6a837aa3-4736-4098-a878-fdeffe227628 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-3.inprogress.f121b73d-ac74-4fbd-b70d-f13e51c9132c (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-4.inprogress.a156884a-f090-4c3f-a271-0b63ab539c45 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-5.inprogress.83c527c5-14dc-4d49-9f99-c915f2224f6a (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/full-v2-template/snapshot (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/only-in-progress-v1/bucket/test-bucket/.part-0-0.inprogress.a70190d6-d080-43a8-b414-746b09d3a8a0 (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/only-in-progress-v1/snapshot (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/only-in-progress-v2/bucket/test-bucket/.part-0-0.inprogress.10833090-dd8c-4e36-884d-bb9758a3a8ef (100%)
copy {flink-streaming-java => flink-connectors/flink-connector-files}/src/test/resources/bucket-state-migration-test/only-in-progress-v2/snapshot (100%)
create mode 100644 flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketStateGenerator.java
copy flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/{utils/NoOpCommitter.java => BucketStatePathResolver.java} (52%)
[flink] 03/10: [FLINK-20337] Let StatefulSinkWriterOperator load
StreamingFileSink's state
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 836d422caa5d497682bae12d74b05910c6863869
Author: GuoWei Ma <gu...@gmail.com>
AuthorDate: Mon Nov 23 12:17:36 2020 +0800
[FLINK-20337] Let StatefulSinkWriterOperator load StreamingFileSink's state
To allow stateful migration from `StreamingFileSink` to `FileSink` we
let the `StatefulSinkWriterOperator` load the `StreamingFileSink`'s
state ("bucket-state") if it exists.
---
.../operators/sink/StatefulSinkWriterOperator.java | 42 +++++++-
.../sink/StatefulSinkWriterOperatorFactory.java | 18 +++-
.../translators/SinkTransformationTranslator.java | 15 ++-
.../sink/StatefulSinkWriterOperatorTest.java | 109 +++++++++++++++++++++
4 files changed, 179 insertions(+), 5 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperator.java
index 3751a3e..65b395c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperator.java
@@ -31,8 +31,13 @@ import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
import org.apache.flink.util.CollectionUtil;
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
import java.util.List;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/**
* Runtime {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing {@link
* SinkWriter Writers} that have state.
@@ -54,18 +59,28 @@ final class StatefulSinkWriterOperator<InputT, CommT, WriterStateT> extends Abst
/** The writer operator's state serializer. */
private final SimpleVersionedSerializer<WriterStateT> writerStateSimpleVersionedSerializer;
+ /** The previous sink operator's state name. */
+ @Nullable
+ private final String previousSinkStateName;
+
// ------------------------------- runtime fields ---------------------------------------
+ /** The previous sink operator's state. */
+ @Nullable
+ private ListState<WriterStateT> previousSinkState;
+
/** The operator's state. */
private ListState<WriterStateT> writerState;
StatefulSinkWriterOperator(
+ @Nullable final String previousSinkStateName,
final ProcessingTimeService processingTimeService,
final Sink<InputT, CommT, WriterStateT, ?> sink,
final SimpleVersionedSerializer<WriterStateT> writerStateSimpleVersionedSerializer) {
super(processingTimeService);
this.sink = sink;
this.writerStateSimpleVersionedSerializer = writerStateSimpleVersionedSerializer;
+ this.previousSinkStateName = previousSinkStateName;
}
@Override
@@ -74,17 +89,40 @@ final class StatefulSinkWriterOperator<InputT, CommT, WriterStateT> extends Abst
final ListState<byte[]> rawState = context.getOperatorStateStore().getListState(WRITER_RAW_STATES_DESC);
writerState = new SimpleVersionedListState<>(rawState, writerStateSimpleVersionedSerializer);
+
+ if (previousSinkStateName != null) {
+ final ListStateDescriptor<byte[]> preSinkStateDesc = new ListStateDescriptor<>(
+ previousSinkStateName,
+ BytePrimitiveArraySerializer.INSTANCE);
+
+ final ListState<byte[]> preRawState = context
+ .getOperatorStateStore()
+ .getListState(preSinkStateDesc);
+ this.previousSinkState = new SimpleVersionedListState<>(
+ preRawState,
+ writerStateSimpleVersionedSerializer);
+ }
}
@SuppressWarnings("unchecked")
@Override
public void snapshotState(StateSnapshotContext context) throws Exception {
writerState.update((List<WriterStateT>) sinkWriter.snapshotState());
+ if (previousSinkState != null) {
+ previousSinkState.clear();
+ }
}
@Override
SinkWriter<InputT, CommT, WriterStateT> createWriter() throws Exception {
- final List<WriterStateT> committables = CollectionUtil.iterableToList(writerState.get());
- return sink.createWriter(createInitContext(), committables);
+ final List<WriterStateT> writerStates = CollectionUtil.iterableToList(writerState.get());
+ final List<WriterStateT> states = new ArrayList<>(writerStates);
+ if (previousSinkStateName != null) {
+ checkNotNull(previousSinkState);
+ final List<WriterStateT> previousSinkStates = CollectionUtil.iterableToList(
+ previousSinkState.get());
+ states.addAll(previousSinkStates);
+ }
+ return sink.createWriter(createInitContext(), states);
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperatorFactory.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperatorFactory.java
index 2741e37..95da487 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperatorFactory.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperatorFactory.java
@@ -23,6 +23,8 @@ import org.apache.flink.api.connector.sink.SinkWriter;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import javax.annotation.Nullable;
+
/**
* A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
* StatefulSinkWriterOperator}.
@@ -35,13 +37,27 @@ public final class StatefulSinkWriterOperatorFactory<InputT, CommT, WriterStateT
private final Sink<InputT, CommT, WriterStateT, ?> sink;
+ @Nullable
+ private final String previousSinkStateName;
+
public StatefulSinkWriterOperatorFactory(Sink<InputT, CommT, WriterStateT, ?> sink) {
+ this(sink, null);
+ }
+
+ public StatefulSinkWriterOperatorFactory(
+ Sink<InputT, CommT, WriterStateT, ?> sink,
+ @Nullable String previousSinkStateName) {
this.sink = sink;
+ this.previousSinkStateName = previousSinkStateName;
}
@Override
AbstractSinkWriterOperator<InputT, CommT> createWriterOperator(ProcessingTimeService processingTimeService) {
- return new StatefulSinkWriterOperator<>(processingTimeService, sink, sink.getWriterStateSerializer().get());
+ return new StatefulSinkWriterOperator<>(
+ previousSinkStateName,
+ processingTimeService,
+ sink,
+ sink.getWriterStateSerializer().get());
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java
index ddd5e06..9e643de 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java
@@ -63,6 +63,9 @@ public class SinkTransformationTranslator<InputT, CommT, WriterStateT, GlobalCom
protected static final Logger LOG = LoggerFactory.getLogger(SinkTransformationTranslator.class);
+ // Currently we only support load the state from streaming file sink;
+ private static final String PREVIOUS_SINK_STATE_NAME = "bucket-states";
+
@Override
public Collection<Integer> translateForBatch(
SinkTransformation<InputT, CommT, WriterStateT, GlobalCommT> transformation,
@@ -75,6 +78,7 @@ public class SinkTransformationTranslator<InputT, CommT, WriterStateT, GlobalCom
internalTranslate(
transformation,
parallelism,
+ PREVIOUS_SINK_STATE_NAME,
new BatchCommitterOperatorFactory<>(transformation.getSink()),
1,
1,
@@ -101,6 +105,7 @@ public class SinkTransformationTranslator<InputT, CommT, WriterStateT, GlobalCom
internalTranslate(
transformation,
parallelism,
+ PREVIOUS_SINK_STATE_NAME,
new StreamingCommitterOperatorFactory<>(transformation.getSink()),
parallelism,
transformation.getMaxParallelism(),
@@ -117,9 +122,9 @@ public class SinkTransformationTranslator<InputT, CommT, WriterStateT, GlobalCom
/**
* Add the sink operators to the stream graph.
- *
* @param sinkTransformation The sink transformation that committer and global committer belongs to.
* @param writerParallelism The parallelism of the writer operator.
+ * @param previousSinkStateName The state name of previous sink's state.
* @param committerFactory The committer operator factory.
* @param committerParallelism The parallelism of the committer operator.
* @param committerMaxParallelism The max parallelism of the committer operator.
@@ -128,6 +133,7 @@ public class SinkTransformationTranslator<InputT, CommT, WriterStateT, GlobalCom
private void internalTranslate(
SinkTransformation<InputT, CommT, WriterStateT, GlobalCommT> sinkTransformation,
int writerParallelism,
+ @SuppressWarnings("SameParameterValue") @Nullable String previousSinkStateName,
OneInputStreamOperatorFactory<CommT, CommT> committerFactory,
int committerParallelism,
int committerMaxParallelism,
@@ -139,6 +145,7 @@ public class SinkTransformationTranslator<InputT, CommT, WriterStateT, GlobalCom
final int writerId = addWriter(
sinkTransformation,
writerParallelism,
+ previousSinkStateName,
context);
final int committerId = addCommitter(
@@ -161,12 +168,14 @@ public class SinkTransformationTranslator<InputT, CommT, WriterStateT, GlobalCom
*
* @param sinkTransformation The transformation that the writer belongs to
* @param parallelism The parallelism of the writer
+ * @param previousSinkStateName The state name of previous sink's state.
*
* @return The stream node id of the writer
*/
private int addWriter(
SinkTransformation<InputT, CommT, WriterStateT, GlobalCommT> sinkTransformation,
int parallelism,
+ @Nullable String previousSinkStateName,
Context context) {
final boolean hasState = sinkTransformation
.getSink()
@@ -180,7 +189,9 @@ public class SinkTransformationTranslator<InputT, CommT, WriterStateT, GlobalCom
final TypeInformation<InputT> inputTypeInfo = input.getOutputType();
final StreamOperatorFactory<CommT> writer =
- hasState ? new StatefulSinkWriterOperatorFactory<>(sinkTransformation.getSink()) : new StatelessSinkWriterOperatorFactory<>(
+ hasState ? new StatefulSinkWriterOperatorFactory<>(
+ sinkTransformation.getSink(),
+ previousSinkStateName) : new StatelessSinkWriterOperatorFactory<>(
sinkTransformation.getSink());
final String prefix = "Sink Writer:";
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperatorTest.java
index bc7c79b..35a93a4 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperatorTest.java
@@ -18,18 +18,31 @@
package org.apache.flink.streaming.runtime.operators.sink;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
import org.apache.flink.api.connector.sink.SinkWriter;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.TestHarnessUtil;
import org.junit.Test;
+import java.util.Arrays;
import java.util.List;
+import java.util.stream.Collectors;
import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.junit.Assert.assertThat;
/**
@@ -90,6 +103,70 @@ public class StatefulSinkWriterOperatorTest extends SinkWriterOperatorTestBase {
new StreamRecord<>(Tuple3.of(2, initialTime + 2, initialTime).toString())));
}
+ @Test
+ public void loadPreviousSinkState() throws Exception {
+ //1. Build previous sink state
+ final List<String> previousSinkInputs = Arrays.asList("bit", "mention", "thick", "stick", "stir",
+ "easy", "sleep", "forth", "cost", "prompt");
+
+ final OneInputStreamOperatorTestHarness<String, String> previousSink =
+ new OneInputStreamOperatorTestHarness<>(
+ new DummySinkOperator(),
+ StringSerializer.INSTANCE);
+
+ OperatorSubtaskState previousSinkState = TestHarnessUtil.buildSubtaskState(
+ previousSink,
+ previousSinkInputs);
+
+ //2. Load previous sink state and verify the output
+ final OneInputStreamOperatorTestHarness<Integer, String> compatibleWriterOperator =
+ createCompatibleSinkOperator();
+
+ final List<StreamRecord<String>> expectedOutput1 =
+ previousSinkInputs.stream().map(StreamRecord::new).collect(Collectors.toList());
+ expectedOutput1.add(new StreamRecord<>(Tuple3.of(1, 1, Long.MIN_VALUE).toString()));
+
+ // load the state from previous sink
+ compatibleWriterOperator.initializeState(previousSinkState);
+
+ compatibleWriterOperator.open();
+
+ compatibleWriterOperator.processElement(1, 1);
+
+ // this will flush out the committables that were restored from previous sink
+ compatibleWriterOperator.endInput();
+
+ OperatorSubtaskState operatorStateWithoutPreviousState = compatibleWriterOperator.snapshot(
+ 1L,
+ 1L);
+
+ compatibleWriterOperator.close();
+
+ assertThat(
+ compatibleWriterOperator.getOutput(),
+ containsInAnyOrder(expectedOutput1.toArray()));
+
+ //3. Restore the sink without previous sink's state
+ final OneInputStreamOperatorTestHarness<Integer, String> restoredSinkOperator =
+ createCompatibleSinkOperator();
+ final List<StreamRecord<String>> expectedOutput2 =
+ Arrays.asList(
+ new StreamRecord<>(Tuple3.of(2, 2, Long.MIN_VALUE).toString()),
+ new StreamRecord<>(Tuple3.of(3, 3, Long.MIN_VALUE).toString()));
+
+ restoredSinkOperator.initializeState(operatorStateWithoutPreviousState);
+
+ restoredSinkOperator.open();
+
+ restoredSinkOperator.processElement(2, 2);
+ restoredSinkOperator.processElement(3, 3);
+
+ // this will flush out the committables that were restored
+ restoredSinkOperator.endInput();
+
+ assertThat(restoredSinkOperator.getOutput(), containsInAnyOrder(expectedOutput2.toArray()));
+ }
+
/**
* A {@link SinkWriter} buffers elements and snapshots them when asked.
*/
@@ -105,4 +182,36 @@ public class StatefulSinkWriterOperatorTest extends SinkWriterOperatorTestBase {
this.elements = states;
}
}
+
+ static class DummySinkOperator extends AbstractStreamOperator<String> implements OneInputStreamOperator<String, String> {
+
+ static final String DUMMY_SINK_STATE_NAME = "dummy_sink_state";
+
+ static final ListStateDescriptor<byte[]> SINK_STATE_DESC = new ListStateDescriptor<>(
+ DUMMY_SINK_STATE_NAME,
+ BytePrimitiveArraySerializer.INSTANCE);
+ ListState<String> sinkState;
+
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sinkState = new SimpleVersionedListState<>(context
+ .getOperatorStateStore()
+ .getListState(SINK_STATE_DESC), TestSink.StringCommittableSerializer.INSTANCE);
+ }
+
+ @Override
+ public void processElement(StreamRecord<String> element) throws Exception {
+ sinkState.add(element.getValue());
+ }
+ }
+
+ private OneInputStreamOperatorTestHarness<Integer, String> createCompatibleSinkOperator() throws Exception {
+ return new OneInputStreamOperatorTestHarness<>(
+ new StatefulSinkWriterOperatorFactory<>(TestSink
+ .newBuilder()
+ .setWriter(new SnapshottingBufferingSinkWriter())
+ .withWriterState()
+ .build(), DummySinkOperator.DUMMY_SINK_STATE_NAME),
+ IntSerializer.INSTANCE);
+ }
}
[flink] 06/10: [refactor] Factor test data generator/resolver out
of BucketStateSerializerTest
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 684bdce974853175d329405c91f31bcaa8d7b733
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Thu Nov 26 12:21:09 2020 +0800
[refactor] Factor test data generator/resolver out of BucketStateSerializerTest
We want to reuse this code for the new FileSink migration test as well.
---
.../sink/filesystem/BucketStateGenerator.java | 173 +++++++++++++++++++++
.../sink/filesystem/BucketStatePathResolver.java | 49 ++++++
.../sink/filesystem/BucketStateSerializerTest.java | 153 ++++--------------
3 files changed, 250 insertions(+), 125 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketStateGenerator.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketStateGenerator.java
new file mode 100644
index 0000000..a35bf55
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketStateGenerator.java
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.functions.sink.filesystem;
+
+import org.apache.flink.api.common.serialization.SimpleStringEncoder;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.io.SimpleVersionedSerialization;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+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.util.FileUtils;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
+/**
+ * Utilities to generate sample bucket states so that we could test migration of bucket states.
+ */
+public class BucketStateGenerator {
+ private final String bucketId;
+ private final String inProgressContent;
+ private final String pendingContent;
+ private final BucketStatePathResolver pathResolver;
+
+ public BucketStateGenerator(
+ String bucketId,
+ String inProgressContent,
+ String pendingContent,
+ java.nio.file.Path basePath,
+ int currentVersion) {
+ this.bucketId = bucketId;
+ this.inProgressContent = inProgressContent;
+ this.pendingContent = pendingContent;
+ this.pathResolver = new BucketStatePathResolver(basePath, currentVersion);
+ }
+
+ public void prepareDeserializationEmpty() throws IOException {
+ final String scenarioName = "empty";
+ final java.nio.file.Path scenarioPath = pathResolver.getResourcePath(scenarioName);
+
+ FileUtils.deleteDirectory(scenarioPath.toFile());
+ Files.createDirectories(scenarioPath);
+
+ final java.nio.file.Path outputPath = pathResolver.getOutputPath(scenarioName);
+ final Path testBucketPath = new Path(outputPath.resolve(bucketId).toString());
+
+ final Bucket<String, String> bucket = createNewBucket(testBucketPath);
+
+ final BucketState<String> bucketState = bucket.onReceptionOfCheckpoint(0);
+
+ byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(
+ bucketStateSerializer(),
+ bucketState);
+ Files.write(pathResolver.getSnapshotPath(scenarioName), bytes);
+ }
+
+ public void prepareDeserializationOnlyInProgress() throws IOException {
+ final String scenarioName = "only-in-progress";
+ final java.nio.file.Path scenarioPath = pathResolver.getResourcePath(scenarioName);
+ FileUtils.deleteDirectory(scenarioPath.toFile());
+ Files.createDirectories(scenarioPath);
+
+ final java.nio.file.Path outputPath = pathResolver.getOutputPath(scenarioName);
+ final Path testBucketPath = new Path(outputPath.resolve(bucketId).toString());
+
+ final Bucket<String, String> bucket =
+ createNewBucket(testBucketPath);
+
+ bucket.write(inProgressContent, System.currentTimeMillis());
+
+ final BucketState<String> bucketState = bucket.onReceptionOfCheckpoint(0);
+
+ final byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(
+ bucketStateSerializer(), bucketState);
+
+ Files.write(pathResolver.getSnapshotPath(scenarioName), bytes);
+ }
+
+ public void prepareDeserializationFull() throws IOException {
+ prepareDeserializationFull(true, "full");
+ }
+
+ public void prepareDeserializationNullInProgress() throws IOException {
+ prepareDeserializationFull(false, "full-no-in-progress");
+ }
+
+ private void prepareDeserializationFull(final boolean withInProgress, final String scenarioName) throws IOException {
+ final java.nio.file.Path scenarioPath = pathResolver.getResourcePath(scenarioName);
+ FileUtils.deleteDirectory(Paths.get(scenarioPath.toString() + "-template").toFile());
+ Files.createDirectories(scenarioPath);
+
+ final int noOfPendingCheckpoints = 5;
+
+ final java.nio.file.Path outputPath = pathResolver.getOutputPath(scenarioName);
+
+ final Path testBucketPath = new Path(outputPath.resolve(bucketId).toString());
+
+ final Bucket<String, String> bucket = createNewBucket(testBucketPath);
+
+ BucketState<String> bucketState = null;
+ // pending for checkpoints
+ for (int i = 0; i < noOfPendingCheckpoints; i++) {
+ // write 10 bytes to the in progress file
+ bucket.write(pendingContent, System.currentTimeMillis());
+ bucket.write(pendingContent, System.currentTimeMillis());
+ // every checkpoint would produce a pending file
+ bucketState = bucket.onReceptionOfCheckpoint(i);
+ }
+
+ if (withInProgress) {
+ // create a in progress file
+ bucket.write(inProgressContent, System.currentTimeMillis());
+
+ // 5 pending files and 1 in progress file
+ bucketState = bucket.onReceptionOfCheckpoint(noOfPendingCheckpoints);
+ }
+
+ final byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(bucketStateSerializer(), bucketState);
+
+ Files.write(pathResolver.getSnapshotPath(scenarioName), bytes);
+
+ // copy the scenario file to a template directory.
+ // it is because that the test `testSerializationFull` would change the in progress file to pending files.
+ moveToTemplateDirectory(scenarioPath);
+ }
+
+ private static RowWiseBucketWriter<String, String> createBucketWriter() throws IOException {
+ return new RowWiseBucketWriter<>(FileSystem.getLocalFileSystem().createRecoverableWriter(), new SimpleStringEncoder<>());
+ }
+
+ private static SimpleVersionedSerializer<BucketState<String>> bucketStateSerializer() throws IOException {
+ final RowWiseBucketWriter bucketWriter = createBucketWriter();
+ return new BucketStateSerializer<>(
+ bucketWriter.getProperties().getInProgressFileRecoverableSerializer(),
+ bucketWriter.getProperties().getPendingFileRecoverableSerializer(),
+ SimpleVersionedStringSerializer.INSTANCE);
+ }
+
+ private Bucket<String, String> createNewBucket(Path bucketPath) throws IOException {
+ return Bucket.getNew(
+ 0,
+ bucketId,
+ bucketPath,
+ 0,
+ createBucketWriter(),
+ DefaultRollingPolicy.builder().withMaxPartSize(10).build(),
+ null,
+ OutputFileConfig.builder().build());
+ }
+
+ private void moveToTemplateDirectory(java.nio.file.Path scenarioPath) throws IOException {
+ FileUtils.copy(new Path(scenarioPath.toString()), new Path(scenarioPath.toString() + "-template"), false);
+ FileUtils.deleteDirectory(scenarioPath.toFile());
+ }
+}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketStatePathResolver.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketStatePathResolver.java
new file mode 100644
index 0000000..1d0f335
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketStatePathResolver.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.functions.sink.filesystem;
+
+import java.nio.file.Path;
+
+/**
+ * Utilities to resolve the directory structure for the bucket state
+ * generated for migration test.
+ */
+public class BucketStatePathResolver {
+ private final java.nio.file.Path basePath;
+ private final int version;
+
+ public BucketStatePathResolver(Path basePath, int version) {
+ this.basePath = basePath;
+ this.version = version;
+ }
+
+ public java.nio.file.Path getSnapshotPath(String scenarioName) {
+ java.nio.file.Path basePath = getResourcePath(scenarioName);
+ return basePath.resolve("snapshot");
+ }
+
+ public java.nio.file.Path getOutputPath(String scenarioName) {
+ java.nio.file.Path basePath = getResourcePath(scenarioName);
+ return basePath.resolve("bucket");
+ }
+
+ public java.nio.file.Path getResourcePath(String scenarioName) {
+ return basePath.resolve(scenarioName + "-v" + version);
+ }
+}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketStateSerializerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketStateSerializerTest.java
index aac6612..3a80257 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketStateSerializerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketStateSerializerTest.java
@@ -76,58 +76,32 @@ public class BucketStateSerializerTest {
private static final String BUCKET_ID = "test-bucket";
+ private static final java.nio.file.Path BASE_PATH = Paths.get("src/test/resources/")
+ .resolve("bucket-state-migration-test");
+
@ClassRule
public static TemporaryFolder tempFolder = new TemporaryFolder();
- private static java.nio.file.Path getResourcePath(
- String scenarioName,
- int version) {
- return Paths.get("src/test/resources/")
- .resolve("bucket-state-migration-test")
- .resolve(scenarioName + "-v" + version);
- }
-
- private static java.nio.file.Path getSnapshotPath(
- String scenarioName,
- int version) {
- java.nio.file.Path basePath = getResourcePath(scenarioName, version);
- return basePath.resolve("snapshot");
- }
-
- private static java.nio.file.Path getOutputPath(String scenarioName, int version) {
- java.nio.file.Path basePath = getResourcePath(scenarioName, version);
- return basePath.resolve("bucket");
- }
+ private final BucketStateGenerator generator = new BucketStateGenerator(
+ BUCKET_ID,
+ IN_PROGRESS_CONTENT,
+ PENDING_CONTENT,
+ BASE_PATH,
+ CURRENT_VERSION);
@Test
@Ignore
public void prepareDeserializationEmpty() throws IOException {
-
- final String scenarioName = "empty";
- final java.nio.file.Path scenarioPath = getResourcePath(scenarioName, CURRENT_VERSION);
-
- FileUtils.deleteDirectory(scenarioPath.toFile());
- Files.createDirectories(scenarioPath);
-
- final java.nio.file.Path outputPath = getOutputPath(scenarioName, CURRENT_VERSION);
- final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
-
- final Bucket<String, String> bucket =
- createNewBucket(testBucketPath);
-
- final BucketState<String> bucketState = bucket.onReceptionOfCheckpoint(0);
-
- byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(
- bucketStateSerializer(),
- bucketState);
- Files.write(getSnapshotPath(scenarioName, CURRENT_VERSION), bytes);
+ generator.prepareDeserializationEmpty();
}
@Test
public void testSerializationEmpty() throws IOException {
final String scenarioName = "empty";
- final java.nio.file.Path outputPath = getOutputPath(scenarioName, previousVersion);
+ final BucketStatePathResolver pathResolver = new BucketStatePathResolver(BASE_PATH, previousVersion);
+
+ final java.nio.file.Path outputPath = pathResolver.getOutputPath(scenarioName);
final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
final BucketState<String> recoveredState = readBucketState(scenarioName, previousVersion);
@@ -141,33 +115,16 @@ public class BucketStateSerializerTest {
@Test
@Ignore
public void prepareDeserializationOnlyInProgress() throws IOException {
-
- final String scenarioName = "only-in-progress";
- final java.nio.file.Path scenarioPath = getResourcePath(scenarioName, CURRENT_VERSION);
- FileUtils.deleteDirectory(scenarioPath.toFile());
- Files.createDirectories(scenarioPath);
-
- final java.nio.file.Path outputPath = getOutputPath(scenarioName, CURRENT_VERSION);
- final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
-
- final Bucket<String, String> bucket =
- createNewBucket(testBucketPath);
-
- bucket.write(IN_PROGRESS_CONTENT, System.currentTimeMillis());
-
- final BucketState<String> bucketState = bucket.onReceptionOfCheckpoint(0);
-
- final byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(
- bucketStateSerializer(), bucketState);
-
- Files.write(getSnapshotPath(scenarioName, CURRENT_VERSION), bytes);
+ generator.prepareDeserializationOnlyInProgress();
}
@Test
public void testSerializationOnlyInProgress() throws IOException {
final String scenarioName = "only-in-progress";
- final java.nio.file.Path outputPath = getOutputPath(scenarioName, previousVersion);
+ final BucketStatePathResolver pathResolver = new BucketStatePathResolver(BASE_PATH, previousVersion);
+
+ final java.nio.file.Path outputPath = pathResolver.getOutputPath(scenarioName);
final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
@@ -195,7 +152,7 @@ public class BucketStateSerializerTest {
@Test
@Ignore
public void prepareDeserializationFull() throws IOException {
- prepareDeserializationFull(true, "full");
+ generator.prepareDeserializationFull();
}
@Test
@@ -206,7 +163,7 @@ public class BucketStateSerializerTest {
@Test
@Ignore
public void prepareDeserializationNullInProgress() throws IOException {
- prepareDeserializationFull(false, "full-no-in-progress");
+ generator.prepareDeserializationNullInProgress();
}
@Test
@@ -214,51 +171,12 @@ public class BucketStateSerializerTest {
testDeserializationFull(false, "full-no-in-progress");
}
- private static void prepareDeserializationFull(final boolean withInProgress, final String scenarioName) throws IOException {
-
- final java.nio.file.Path scenarioPath = getResourcePath(scenarioName, CURRENT_VERSION);
- FileUtils.deleteDirectory(Paths.get(scenarioPath.toString() + "-template").toFile());
- Files.createDirectories(scenarioPath);
-
- final int noOfPendingCheckpoints = 5;
-
- final java.nio.file.Path outputPath = getOutputPath(scenarioName, CURRENT_VERSION);
-
- final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
-
- final Bucket<String, String> bucket = createNewBucket(testBucketPath);
-
- BucketState<String> bucketState = null;
- // pending for checkpoints
- for (int i = 0; i < noOfPendingCheckpoints; i++) {
- // write 10 bytes to the in progress file
- bucket.write(PENDING_CONTENT, System.currentTimeMillis());
- bucket.write(PENDING_CONTENT, System.currentTimeMillis());
- // every checkpoint would produce a pending file
- bucketState = bucket.onReceptionOfCheckpoint(i);
- }
-
- if (withInProgress) {
- // create a in progress file
- bucket.write(IN_PROGRESS_CONTENT, System.currentTimeMillis());
-
- // 5 pending files and 1 in progress file
- bucketState = bucket.onReceptionOfCheckpoint(noOfPendingCheckpoints);
- }
-
- final byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(bucketStateSerializer(), bucketState);
-
- Files.write(getSnapshotPath(scenarioName, CURRENT_VERSION), bytes);
-
- // copy the scenario file to a template directory.
- // it is because that the test `testSerializationFull` would change the in progress file to pending files.
- moveToTemplateDirectory(scenarioPath);
- }
-
private void testDeserializationFull(final boolean withInProgress, final String scenarioName) throws IOException {
+ final BucketStatePathResolver pathResolver = new BucketStatePathResolver(BASE_PATH, previousVersion);
+
try {
- final java.nio.file.Path outputPath = getOutputPath(scenarioName, previousVersion);
+ final java.nio.file.Path outputPath = pathResolver.getOutputPath(scenarioName);
final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
// restore the state
final BucketState<String> recoveredState = readBucketStateFromTemplate(scenarioName, previousVersion);
@@ -306,22 +224,10 @@ public class BucketStateSerializerTest {
assertThat(afterRestorePaths, empty());
}
} finally {
- FileUtils.deleteDirectory(getResourcePath(scenarioName, previousVersion).toFile());
+ FileUtils.deleteDirectory(pathResolver.getResourcePath(scenarioName).toFile());
}
}
- private static Bucket<String, String> createNewBucket(final Path bucketPath) throws IOException {
- return Bucket.getNew(
- 0,
- BUCKET_ID,
- bucketPath,
- 0,
- createBucketWriter(),
- DefaultRollingPolicy.builder().withMaxPartSize(10).build(),
- null,
- OutputFileConfig.builder().build());
- }
-
private static Bucket<String, String> restoreBucket(final int initialPartCounter, final BucketState<String> bucketState) throws IOException {
return Bucket.restore(
0,
@@ -338,7 +244,7 @@ public class BucketStateSerializerTest {
}
private static SimpleVersionedSerializer<BucketState<String>> bucketStateSerializer() throws IOException {
- final RowWiseBucketWriter bucketWriter = createBucketWriter();
+ final RowWiseBucketWriter<String, String> bucketWriter = createBucketWriter();
return new BucketStateSerializer<>(
bucketWriter.getProperties().getInProgressFileRecoverableSerializer(),
bucketWriter.getProperties().getPendingFileRecoverableSerializer(),
@@ -346,12 +252,14 @@ public class BucketStateSerializerTest {
}
private static BucketState<String> readBucketState(final String scenarioName, final int version) throws IOException {
- byte[] bytes = Files.readAllBytes(getSnapshotPath(scenarioName, version));
+ final BucketStatePathResolver pathResolver = new BucketStatePathResolver(BASE_PATH, version);
+ byte[] bytes = Files.readAllBytes(pathResolver.getSnapshotPath(scenarioName));
return SimpleVersionedSerialization.readVersionAndDeSerialize(bucketStateSerializer(), bytes);
}
private static BucketState<String> readBucketStateFromTemplate(final String scenarioName, final int version) throws IOException {
- final java.nio.file.Path scenarioPath = getResourcePath(scenarioName, version);
+ final BucketStatePathResolver pathResolver = new BucketStatePathResolver(BASE_PATH, version);
+ final java.nio.file.Path scenarioPath = pathResolver.getResourcePath(scenarioName);
// clear the scenario files first
FileUtils.deleteDirectory(scenarioPath.toFile());
@@ -361,9 +269,4 @@ public class BucketStateSerializerTest {
return readBucketState(scenarioName, version);
}
-
- private static void moveToTemplateDirectory(java.nio.file.Path scenarioPath) throws IOException {
- FileUtils.copy(new Path(scenarioPath.toString()), new Path(scenarioPath.toString() + "-template"), false);
- FileUtils.deleteDirectory(scenarioPath.toFile());
- }
}
[flink] 10/10: [refactor] Move FileSink migration tests to
flink-connector-files
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 56bfa3332dbeacfe50ac9fe2835f0c22c9b30170
Author: Aljoscha Krettek <al...@apache.org>
AuthorDate: Mon Nov 30 13:15:59 2020 +0100
[refactor] Move FileSink migration tests to flink-connector-files
---
flink-connectors/flink-connector-files/pom.xml | 15 +++++
.../file/sink/writer/FileSinkMigrationITCase.java | 2 +-
...leWriterBucketStateSerializerMigrationTest.java | 2 +-
.../bucket-state-migration-test/empty-v1/snapshot | Bin
.../bucket-state-migration-test/empty-v2/snapshot | Bin
...inprogress.a88d5993-77bc-44ce-880b-9f2a43b59ab4 | 0
...inprogress.7c0f2bd7-3078-48e8-9af2-d8773fb949c5 | 0
...inprogress.6729a640-0585-4785-a652-89802950c663 | 0
...inprogress.b4bcb0e9-5c9e-45dd-8963-1b163343544d | 0
...inprogress.e1e9e48d-0db6-4dd7-8a4d-fb4ebe7ed8ac | 0
.../full-no-in-progress-v1-template/snapshot | Bin
...inprogress.1e22e72d-0ab2-493b-8b00-9edac4252cec | 0
...inprogress.3821f491-9fa1-48b2-b66b-655352a3c8ec | 0
...inprogress.0af18f41-d8f8-4a4e-a92e-de12851be20b | 0
...inprogress.a3d0f4d2-d6ad-4f83-ba62-ed4b1fa86db2 | 0
...inprogress.666acf3e-935c-4621-8171-f7c897496524 | 0
.../full-no-in-progress-v2-template/snapshot | Bin
...inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82 | 0
...inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c | 0
...inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97 | 0
...inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9 | 0
...inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7 | 0
...inprogress.32f5a28f-20e1-48da-9951-10e795133d64 | 0
.../full-v1-template/snapshot | Bin
...inprogress.9731063e-2b28-4701-8cc1-e706480b8022 | 0
...inprogress.1d423406-097a-4deb-bfde-d023d3477cd5 | 0
...inprogress.6a837aa3-4736-4098-a878-fdeffe227628 | 0
...inprogress.f121b73d-ac74-4fbd-b70d-f13e51c9132c | 0
...inprogress.a156884a-f090-4c3f-a271-0b63ab539c45 | 0
...inprogress.83c527c5-14dc-4d49-9f99-c915f2224f6a | 0
.../full-v2-template/snapshot | Bin
...inprogress.a70190d6-d080-43a8-b414-746b09d3a8a0 | 0
.../only-in-progress-v1/snapshot | Bin
...inprogress.10833090-dd8c-4e36-884d-bb9758a3a8ef | 0
.../only-in-progress-v2/snapshot | Bin
.../flink-connector-migration-test/pom.xml | 66 ---------------------
.../src/test/resources/log4j2-test.properties | 29 ---------
flink-connectors/pom.xml | 1 -
38 files changed, 17 insertions(+), 98 deletions(-)
diff --git a/flink-connectors/flink-connector-files/pom.xml b/flink-connectors/flink-connector-files/pom.xml
index 65a9027..7fec53c 100644
--- a/flink-connectors/flink-connector-files/pom.xml
+++ b/flink-connectors/flink-connector-files/pom.xml
@@ -90,6 +90,21 @@ under the License.
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
<build>
diff --git a/flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileSinkMigrationITCase.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileSinkMigrationITCase.java
similarity index 99%
rename from flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileSinkMigrationITCase.java
rename to flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileSinkMigrationITCase.java
index 2f6ed68..09e2252 100644
--- a/flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileSinkMigrationITCase.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileSinkMigrationITCase.java
@@ -7,7 +7,7 @@
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
diff --git a/flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerMigrationTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerMigrationTest.java
similarity index 99%
rename from flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerMigrationTest.java
rename to flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerMigrationTest.java
index 4cce402..4715cd6 100644
--- a/flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerMigrationTest.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerMigrationTest.java
@@ -7,7 +7,7 @@
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/empty-v1/snapshot b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/empty-v1/snapshot
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/empty-v1/snapshot
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/empty-v1/snapshot
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/empty-v2/snapshot b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/empty-v2/snapshot
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/empty-v2/snapshot
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/empty-v2/snapshot
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-0.inprogress.a88d5993-77bc-44ce-880b-9f2a43b59ab4 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-0.inprogress.a88d5993-77bc-44ce-880b-9f2a43b59ab4
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-0.inprogress.a88d5993-77bc-44ce-880b-9f2a43b59ab4
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-0.inprogress.a88d5993-77bc-44ce-880b-9f2a43b59ab4
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-1.inprogress.7c0f2bd7-3078-48e8-9af2-d8773fb949c5 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-1.inprogress.7c0f2bd7-3078-48e8-9af2-d8773fb949c5
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-1.inprogress.7c0f2bd7-3078-48e8-9af2-d8773fb949c5
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-1.inprogress.7c0f2bd7-3078-48e8-9af2-d8773fb949c5
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-2.inprogress.6729a640-0585-4785-a652-89802950c663 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-2.inprogress.6729a640-0585-4785-a652-89802950c663
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-2.inprogress.6729a640-0585-4785-a652-89802950c663
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-2.inprogress.6729a640-0585-4785-a652-89802950c663
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-3.inprogress.b4bcb0e9-5c9e-45dd-8963-1b163343544d b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-3.inprogress.b4bcb0e9-5c9e-45dd-8963-1b163343544d
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-3.inprogress.b4bcb0e9-5c9e-45dd-8963-1b163343544d
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-3.inprogress.b4bcb0e9-5c9e-45dd-8963-1b163343544d
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-4.inprogress.e1e9e48d-0db6-4dd7-8a4d-fb4ebe7ed8ac b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-4.inprogress.e1e9e48d-0db6-4dd7-8a4d-fb4ebe7ed8ac
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-4.inprogress.e1e9e48d-0db6-4dd7-8a4d-fb4ebe7ed8ac
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-4.inprogress.e1e9e48d-0db6-4dd7-8a4d-fb4ebe7ed8ac
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/snapshot b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/snapshot
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/snapshot
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/snapshot
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-0.inprogress.1e22e72d-0ab2-493b-8b00-9edac4252cec b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-0.inprogress.1e22e72d-0ab2-493b-8b00-9edac4252cec
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-0.inprogress.1e22e72d-0ab2-493b-8b00-9edac4252cec
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-0.inprogress.1e22e72d-0ab2-493b-8b00-9edac4252cec
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-1.inprogress.3821f491-9fa1-48b2-b66b-655352a3c8ec b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-1.inprogress.3821f491-9fa1-48b2-b66b-655352a3c8ec
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-1.inprogress.3821f491-9fa1-48b2-b66b-655352a3c8ec
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-1.inprogress.3821f491-9fa1-48b2-b66b-655352a3c8ec
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-2.inprogress.0af18f41-d8f8-4a4e-a92e-de12851be20b b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-2.inprogress.0af18f41-d8f8-4a4e-a92e-de12851be20b
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-2.inprogress.0af18f41-d8f8-4a4e-a92e-de12851be20b
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-2.inprogress.0af18f41-d8f8-4a4e-a92e-de12851be20b
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-3.inprogress.a3d0f4d2-d6ad-4f83-ba62-ed4b1fa86db2 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-3.inprogress.a3d0f4d2-d6ad-4f83-ba62-ed4b1fa86db2
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-3.inprogress.a3d0f4d2-d6ad-4f83-ba62-ed4b1fa86db2
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-3.inprogress.a3d0f4d2-d6ad-4f83-ba62-ed4b1fa86db2
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-4.inprogress.666acf3e-935c-4621-8171-f7c897496524 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-4.inprogress.666acf3e-935c-4621-8171-f7c897496524
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-4.inprogress.666acf3e-935c-4621-8171-f7c897496524
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-4.inprogress.666acf3e-935c-4621-8171-f7c897496524
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/snapshot b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/snapshot
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/snapshot
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/snapshot
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-0.inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-0.inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-0.inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-0.inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-1.inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-1.inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-1.inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-1.inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-2.inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-2.inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-2.inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-2.inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-3.inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-3.inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-3.inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-3.inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-4.inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-4.inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-4.inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-4.inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-5.inprogress.32f5a28f-20e1-48da-9951-10e795133d64 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-5.inprogress.32f5a28f-20e1-48da-9951-10e795133d64
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-5.inprogress.32f5a28f-20e1-48da-9951-10e795133d64
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-5.inprogress.32f5a28f-20e1-48da-9951-10e795133d64
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/snapshot b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/snapshot
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/snapshot
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v1-template/snapshot
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-0.inprogress.9731063e-2b28-4701-8cc1-e706480b8022 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-0.inprogress.9731063e-2b28-4701-8cc1-e706480b8022
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-0.inprogress.9731063e-2b28-4701-8cc1-e706480b8022
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-0.inprogress.9731063e-2b28-4701-8cc1-e706480b8022
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-1.inprogress.1d423406-097a-4deb-bfde-d023d3477cd5 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-1.inprogress.1d423406-097a-4deb-bfde-d023d3477cd5
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-1.inprogress.1d423406-097a-4deb-bfde-d023d3477cd5
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-1.inprogress.1d423406-097a-4deb-bfde-d023d3477cd5
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-2.inprogress.6a837aa3-4736-4098-a878-fdeffe227628 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-2.inprogress.6a837aa3-4736-4098-a878-fdeffe227628
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-2.inprogress.6a837aa3-4736-4098-a878-fdeffe227628
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-2.inprogress.6a837aa3-4736-4098-a878-fdeffe227628
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-3.inprogress.f121b73d-ac74-4fbd-b70d-f13e51c9132c b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-3.inprogress.f121b73d-ac74-4fbd-b70d-f13e51c9132c
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-3.inprogress.f121b73d-ac74-4fbd-b70d-f13e51c9132c
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-3.inprogress.f121b73d-ac74-4fbd-b70d-f13e51c9132c
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-4.inprogress.a156884a-f090-4c3f-a271-0b63ab539c45 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-4.inprogress.a156884a-f090-4c3f-a271-0b63ab539c45
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-4.inprogress.a156884a-f090-4c3f-a271-0b63ab539c45
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-4.inprogress.a156884a-f090-4c3f-a271-0b63ab539c45
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-5.inprogress.83c527c5-14dc-4d49-9f99-c915f2224f6a b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-5.inprogress.83c527c5-14dc-4d49-9f99-c915f2224f6a
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-5.inprogress.83c527c5-14dc-4d49-9f99-c915f2224f6a
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-5.inprogress.83c527c5-14dc-4d49-9f99-c915f2224f6a
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/snapshot b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/snapshot
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/snapshot
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/full-v2-template/snapshot
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v1/bucket/test-bucket/.part-0-0.inprogress.a70190d6-d080-43a8-b414-746b09d3a8a0 b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/only-in-progress-v1/bucket/test-bucket/.part-0-0.inprogress.a70190d6-d080-43a8-b414-746b09d3a8a0
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v1/bucket/test-bucket/.part-0-0.inprogress.a70190d6-d080-43a8-b414-746b09d3a8a0
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/only-in-progress-v1/bucket/test-bucket/.part-0-0.inprogress.a70190d6-d080-43a8-b414-746b09d3a8a0
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v1/snapshot b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/only-in-progress-v1/snapshot
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v1/snapshot
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/only-in-progress-v1/snapshot
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v2/bucket/test-bucket/.part-0-0.inprogress.10833090-dd8c-4e36-884d-bb9758a3a8ef b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/only-in-progress-v2/bucket/test-bucket/.part-0-0.inprogress.10833090-dd8c-4e36-884d-bb9758a3a8ef
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v2/bucket/test-bucket/.part-0-0.inprogress.10833090-dd8c-4e36-884d-bb9758a3a8ef
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/only-in-progress-v2/bucket/test-bucket/.part-0-0.inprogress.10833090-dd8c-4e36-884d-bb9758a3a8ef
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v2/snapshot b/flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/only-in-progress-v2/snapshot
similarity index 100%
rename from flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v2/snapshot
rename to flink-connectors/flink-connector-files/src/test/resources/bucket-state-migration-test/only-in-progress-v2/snapshot
diff --git a/flink-connectors/flink-connector-migration-test/pom.xml b/flink-connectors/flink-connector-migration-test/pom.xml
deleted file mode 100644
index 55b0f41..0000000
--- a/flink-connectors/flink-connector-migration-test/pom.xml
+++ /dev/null
@@ -1,66 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied. See the License for the
-specific language governing permissions and limitations
-under the License.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
-
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connectors</artifactId>
- <version>1.12-SNAPSHOT</version>
- <relativePath>..</relativePath>
- </parent>
-
- <artifactId>flink-file-migration-test</artifactId>
- <name>Flink : Connectors : Migration Test</name>
-
- <packaging>jar</packaging>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-files</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-files</artifactId>
- <version>${project.version}</version>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
- </dependencies>
-</project>
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/log4j2-test.properties b/flink-connectors/flink-connector-migration-test/src/test/resources/log4j2-test.properties
deleted file mode 100644
index 6faab06..0000000
--- a/flink-connectors/flink-connector-migration-test/src/test/resources/log4j2-test.properties
+++ /dev/null
@@ -1,29 +0,0 @@
-################################################################################
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-################################################################################
-
-# Set root logger level to OFF to not flood build logs
-# set manually to INFO for debugging purposes
-rootLogger.level = OFF
-rootLogger.appenderRef.test.ref = TestLogger
-
-appender.testlogger.name = TestLogger
-appender.testlogger.type = CONSOLE
-appender.testlogger.target = SYSTEM_ERR
-appender.testlogger.layout.type = PatternLayout
-appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
-
diff --git a/flink-connectors/pom.xml b/flink-connectors/pom.xml
index 1f91f6c..11ef77e 100644
--- a/flink-connectors/pom.xml
+++ b/flink-connectors/pom.xml
@@ -57,7 +57,6 @@ under the License.
<module>flink-connector-base</module>
<module>flink-file-sink-common</module>
<module>flink-connector-files</module>
- <module>flink-connector-migration-test</module>
</modules>
<!-- override these root dependencies as 'provided', so they don't end up
[flink] 09/10: [FLINK-20337] Add ITCase for migrating from
StreamingFileSink to FileSink
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c54f5a766505ea64e0c590ab68e0f121ef3ebc73
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Fri Nov 27 19:33:15 2020 +0800
[FLINK-20337] Add ITCase for migrating from StreamingFileSink to FileSink
---
.../flink-connector-migration-test/pom.xml | 8 +
.../file/sink/writer/FileSinkMigrationITCase.java | 287 +++++++++++++++++++++
2 files changed, 295 insertions(+)
diff --git a/flink-connectors/flink-connector-migration-test/pom.xml b/flink-connectors/flink-connector-migration-test/pom.xml
index 9a16ada..55b0f41 100644
--- a/flink-connectors/flink-connector-migration-test/pom.xml
+++ b/flink-connectors/flink-connector-migration-test/pom.xml
@@ -54,5 +54,13 @@ under the License.
<type>test-jar</type>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-files</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileSinkMigrationITCase.java b/flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileSinkMigrationITCase.java
new file mode 100644
index 0000000..2f6ed68
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileSinkMigrationITCase.java
@@ -0,0 +1,287 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.sink.writer;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.api.common.state.CheckpointListener;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.connector.file.sink.FileSink;
+import org.apache.flink.connector.file.sink.utils.IntegerFileSinkTestDataUtils;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
+import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * Tests migrating from {@link StreamingFileSink} to {@link FileSink}. It trigger a savepoint
+ * for the {@link StreamingFileSink} job and restore the {@link FileSink} job from the savepoint
+ * taken.
+ */
+public class FileSinkMigrationITCase extends TestLogger {
+
+ @ClassRule
+ public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
+
+ private static final String SOURCE_UID = "source";
+
+ private static final String SINK_UID = "sink";
+
+ private static final int NUM_SOURCES = 4;
+
+ private static final int NUM_SINKS = 3;
+
+ private static final int NUM_RECORDS = 10000;
+
+ private static final int NUM_BUCKETS = 4;
+
+ private static final Map<String, CountDownLatch> SAVEPOINT_LATCH_MAP = new ConcurrentHashMap<>();
+
+ private static final Map<String, CountDownLatch> FINAL_CHECKPOINT_LATCH_MAP = new ConcurrentHashMap<>();
+
+ private String latchId;
+
+ @Before
+ public void setup() {
+ this.latchId = UUID.randomUUID().toString();
+ SAVEPOINT_LATCH_MAP.put(latchId, new CountDownLatch(NUM_SOURCES));
+
+ // We wait for two successful checkpoints in sources before shutting down. This ensures that
+ // the sink can commit its data.
+ // We need to keep a "static" latch here because all sources need to be kept running
+ // while we're waiting for the required number of checkpoints. Otherwise, we would lock up
+ // because we can only do checkpoints while all operators are running.
+ FINAL_CHECKPOINT_LATCH_MAP.put(latchId, new CountDownLatch(NUM_SOURCES * 2));
+ }
+
+ @After
+ public void teardown() {
+ SAVEPOINT_LATCH_MAP.remove(latchId);
+ FINAL_CHECKPOINT_LATCH_MAP.remove(latchId);
+ }
+
+ @Test
+ public void testMigration() throws Exception {
+ String outputPath = TEMPORARY_FOLDER.newFolder().getAbsolutePath();
+ String savepointBasePath = TEMPORARY_FOLDER.newFolder().getAbsolutePath();
+
+ final Configuration config = new Configuration();
+ config.setString(RestOptions.BIND_PORT, "18081-19000");
+ final MiniClusterConfiguration cfg = new MiniClusterConfiguration.Builder()
+ .setNumTaskManagers(1)
+ .setNumSlotsPerTaskManager(4)
+ .setConfiguration(config)
+ .build();
+
+ JobGraph streamingFileSinkJobGraph = createStreamingFileSinkJobGraph(outputPath);
+ String savepointPath = executeAndTakeSavepoint(cfg, streamingFileSinkJobGraph, savepointBasePath);
+
+ JobGraph fileSinkJobGraph = createFileSinkJobGraph(outputPath);
+ loadSavepointAndExecute(cfg, fileSinkJobGraph, savepointPath);
+
+ IntegerFileSinkTestDataUtils.checkIntegerSequenceSinkOutput(outputPath, NUM_RECORDS, NUM_BUCKETS, NUM_SOURCES);
+ }
+
+ private JobGraph createStreamingFileSinkJobGraph(String outputPath) {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.enableCheckpointing(500, CheckpointingMode.EXACTLY_ONCE);
+
+ StreamingFileSink<Integer> sink = StreamingFileSink
+ .forRowFormat(new Path(outputPath), new IntegerFileSinkTestDataUtils.IntEncoder())
+ .withBucketAssigner(new IntegerFileSinkTestDataUtils.ModuloBucketAssigner(NUM_BUCKETS))
+ .withRollingPolicy(OnCheckpointRollingPolicy.build())
+ .build();
+
+ env.addSource(new StatefulSource(true, latchId))
+ .uid(SOURCE_UID)
+ .setParallelism(NUM_SOURCES)
+ .addSink(sink)
+ .setParallelism(NUM_SINKS)
+ .uid(SINK_UID);
+ return env.getStreamGraph().getJobGraph();
+ }
+
+ private JobGraph createFileSinkJobGraph(String outputPath) {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.enableCheckpointing(500, CheckpointingMode.EXACTLY_ONCE);
+
+ FileSink<Integer> sink = FileSink
+ .forRowFormat(new Path(outputPath), new IntegerFileSinkTestDataUtils.IntEncoder())
+ .withBucketAssigner(new IntegerFileSinkTestDataUtils.ModuloBucketAssigner(NUM_BUCKETS))
+ .withRollingPolicy(OnCheckpointRollingPolicy.build())
+ .build();
+
+ env.addSource(new StatefulSource(false, latchId))
+ .uid(SOURCE_UID)
+ .setParallelism(NUM_SOURCES)
+ .sinkTo(sink)
+ .setParallelism(NUM_SINKS)
+ .uid(SINK_UID);
+ return env.getStreamGraph().getJobGraph();
+ }
+
+ private String executeAndTakeSavepoint(
+ MiniClusterConfiguration cfg,
+ JobGraph jobGraph,
+ String savepointBasePath) throws Exception {
+ try (MiniCluster miniCluster = new MiniCluster(cfg)) {
+ miniCluster.start();
+ CompletableFuture<JobSubmissionResult> jobSubmissionResultFuture = miniCluster.submitJob(jobGraph);
+ JobID jobId = jobSubmissionResultFuture.get().getJobID();
+
+ // wait till we can taking savepoint
+ CountDownLatch latch = SAVEPOINT_LATCH_MAP.get(latchId);
+ latch.await();
+
+ CompletableFuture<String> savepointResultFuture = miniCluster.triggerSavepoint(jobId, savepointBasePath, true);
+ return savepointResultFuture.get();
+ }
+ }
+
+ private void loadSavepointAndExecute(
+ MiniClusterConfiguration cfg,
+ JobGraph jobGraph,
+ String savepointPath
+ ) throws Exception {
+ jobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(savepointPath, false));
+
+ try (MiniCluster miniCluster = new MiniCluster(cfg)) {
+ miniCluster.start();
+ miniCluster.executeJobBlocking(jobGraph);
+ }
+ }
+
+ private static class StatefulSource extends RichParallelSourceFunction<Integer> implements
+ CheckpointedFunction, CheckpointListener {
+
+ private final boolean takingSavepointMode;
+
+ private final String latchId;
+
+ private ListState<Integer> nextValueState;
+
+ private int nextValue;
+
+ private volatile boolean snapshottedAfterAllRecordsOutput;
+
+ private volatile boolean isWaitingCheckpointComplete;
+
+ private volatile boolean isCanceled;
+
+ public StatefulSource(boolean takingSavepointMode, String latchId) {
+ this.takingSavepointMode = takingSavepointMode;
+ this.latchId = latchId;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ nextValueState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>("nextValue", Integer.class));
+
+ if (nextValueState.get() != null && nextValueState.get().iterator().hasNext()) {
+ nextValue = nextValueState.get().iterator().next();
+ }
+ }
+
+ @Override
+ public void run(SourceContext<Integer> ctx) throws Exception {
+ if (takingSavepointMode) {
+ sendRecordsUntil(NUM_RECORDS / 3, 0, ctx);
+
+ CountDownLatch latch = SAVEPOINT_LATCH_MAP.get(latchId);
+ latch.countDown();
+
+ sendRecordsUntil(NUM_RECORDS / 2, 100, ctx);
+
+ while (true) {
+ Thread.sleep(5000);
+ }
+ } else {
+ sendRecordsUntil(NUM_RECORDS, 0, ctx);
+
+ // Wait the last checkpoint to commit all the pending records.
+ isWaitingCheckpointComplete = true;
+ CountDownLatch latch = FINAL_CHECKPOINT_LATCH_MAP.get(latchId);
+ latch.await();
+ }
+ }
+
+ private void sendRecordsUntil(int targetNumber, int sleepInMillis, SourceContext<Integer> ctx)
+ throws InterruptedException {
+ while (!isCanceled && nextValue < targetNumber) {
+ synchronized (ctx.getCheckpointLock()) {
+ ctx.collect(nextValue++);
+ }
+
+ if (sleepInMillis > 0) {
+ Thread.sleep(sleepInMillis);
+ }
+ }
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ nextValueState.update(Collections.singletonList(nextValue));
+
+ if (isWaitingCheckpointComplete) {
+ snapshottedAfterAllRecordsOutput = true;
+ }
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ if (isWaitingCheckpointComplete && snapshottedAfterAllRecordsOutput) {
+ CountDownLatch latch = FINAL_CHECKPOINT_LATCH_MAP.get(latchId);
+ latch.countDown();
+ }
+ }
+
+ @Override
+ public void cancel() {
+ isCanceled = true;
+ }
+
+ }
+}
[flink] 05/10: [FLINK-20337] Let FileSink restore state from
StreamingFileSink
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d0757e95c3c33d4f13377f2c14c179076b3795b2
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Wed Nov 25 23:29:38 2020 +0800
[FLINK-20337] Let FileSink restore state from StreamingFileSink
The FileWriterBucketStateSerializer is bumped past version 2 to allow it
to restore the same old versions as BucketStateSerialize (from
StreamingFileSink).
---
.../apache/flink/connector/file/sink/FileSink.java | 14 +--
.../file/sink/writer/FileWriterBucket.java | 12 +++
.../file/sink/writer/FileWriterBucketState.java | 33 +++++++
.../writer/FileWriterBucketStateSerializer.java | 110 ++++++++++++++++++---
.../FileWriterBucketStateSerializerTest.java | 4 +-
.../file/sink/writer/FileWriterBucketTest.java | 59 +++++++++++
.../OutputStreamBasedPartFileWriter.java | 23 +++--
7 files changed, 229 insertions(+), 26 deletions(-)
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java
index 13d36d3..ccd010d 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java
@@ -293,10 +293,11 @@ public class FileSink<IN>
@Override
SimpleVersionedSerializer<FileWriterBucketState> getWriterStateSerializer() throws IOException {
+ BucketWriter<IN, String> bucketWriter = createBucketWriter();
+
return new FileWriterBucketStateSerializer(
- createBucketWriter()
- .getProperties()
- .getInProgressFileRecoverableSerializer());
+ bucketWriter.getProperties().getInProgressFileRecoverableSerializer(),
+ bucketWriter.getProperties().getPendingFileRecoverableSerializer());
}
@Override
@@ -444,10 +445,11 @@ public class FileSink<IN>
@Override
SimpleVersionedSerializer<FileWriterBucketState> getWriterStateSerializer() throws IOException {
+ BucketWriter<IN, String> bucketWriter = createBucketWriter();
+
return new FileWriterBucketStateSerializer(
- createBucketWriter()
- .getProperties()
- .getInProgressFileRecoverableSerializer());
+ bucketWriter.getProperties().getInProgressFileRecoverableSerializer(),
+ bucketWriter.getProperties().getPendingFileRecoverableSerializer());
}
@Override
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucket.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucket.java
index 33a4469..5cb135a 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucket.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucket.java
@@ -118,6 +118,10 @@ class FileWriterBucket<IN> {
outputFileConfig);
restoreInProgressFile(bucketState);
+
+ // Restore pending files, this only make difference if we are
+ // migrating from {@code StreamingFileSink}.
+ cacheRecoveredPendingFiles(bucketState);
}
private void restoreInProgressFile(FileWriterBucketState state) throws IOException {
@@ -137,6 +141,14 @@ class FileWriterBucket<IN> {
}
}
+ private void cacheRecoveredPendingFiles(FileWriterBucketState state) {
+ // Cache the previous pending files and send to committer on the first prepareCommit operation.
+ for (List<InProgressFileWriter.PendingFileRecoverable> restoredPendingRecoverables :
+ state.getPendingFileRecoverablesPerCheckpoint().values()) {
+ pendingFiles.addAll(restoredPendingRecoverables);
+ }
+ }
+
public String getBucketId() {
return bucketId;
}
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketState.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketState.java
index 72f88c1..cbab95e 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketState.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketState.java
@@ -23,7 +23,12 @@ import org.apache.flink.core.fs.Path;
import javax.annotation.Nullable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
import static org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter.InProgressFileRecoverable;
+import static org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter.PendingFileRecoverable;
/**
* States for {@link FileWriterBucket}.
@@ -49,15 +54,31 @@ public class FileWriterBucketState {
@Nullable
private final InProgressFileRecoverable inProgressFileRecoverable;
+ /**
+ * The {@link PendingFileRecoverable} should be empty unless we
+ * are migrating from {@code StreamingFileSink}.
+ */
+ private final Map<Long, List<PendingFileRecoverable>> pendingFileRecoverablesPerCheckpoint;
+
public FileWriterBucketState(
String bucketId,
Path bucketPath,
long inProgressFileCreationTime,
@Nullable InProgressFileRecoverable inProgressFileRecoverable) {
+ this(bucketId, bucketPath, inProgressFileCreationTime, inProgressFileRecoverable, new HashMap<>());
+ }
+
+ public FileWriterBucketState(
+ String bucketId,
+ Path bucketPath,
+ long inProgressFileCreationTime,
+ @Nullable InProgressFileRecoverable inProgressFileRecoverable,
+ Map<Long, List<PendingFileRecoverable>> pendingFileRecoverablesPerCheckpoint) {
this.bucketId = bucketId;
this.bucketPath = bucketPath;
this.inProgressFileCreationTime = inProgressFileCreationTime;
this.inProgressFileRecoverable = inProgressFileRecoverable;
+ this.pendingFileRecoverablesPerCheckpoint = pendingFileRecoverablesPerCheckpoint;
}
public String getBucketId() {
@@ -81,6 +102,10 @@ public class FileWriterBucketState {
return inProgressFileRecoverable != null;
}
+ public Map<Long, List<PendingFileRecoverable>> getPendingFileRecoverablesPerCheckpoint() {
+ return pendingFileRecoverablesPerCheckpoint;
+ }
+
@Override
public String toString() {
final StringBuilder strBuilder = new StringBuilder();
@@ -92,6 +117,14 @@ public class FileWriterBucketState {
if (hasInProgressFileRecoverable()) {
strBuilder.append(", has open part file created @ ").append(inProgressFileCreationTime);
}
+
+ if (!pendingFileRecoverablesPerCheckpoint.isEmpty()) {
+ strBuilder.append(", has pending files for checkpoints: {");
+ for (long checkpointId: pendingFileRecoverablesPerCheckpoint.keySet()) {
+ strBuilder.append(checkpointId).append(' ');
+ }
+ strBuilder.append('}');
+ }
return strBuilder.toString();
}
}
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializer.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializer.java
index a40cddb9..659ee0d 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializer.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializer.java
@@ -20,17 +20,28 @@ package org.apache.flink.connector.file.sink.writer;
import org.apache.flink.annotation.Internal;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.fs.RecoverableWriter;
import org.apache.flink.core.io.SimpleVersionedSerialization;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter;
+import org.apache.flink.streaming.api.functions.sink.filesystem.OutputStreamBasedPartFileWriter;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
+import org.apache.flink.util.function.BiFunctionWithException;
+import org.apache.flink.util.function.FunctionWithException;
+
+import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
import static org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter.InProgressFileRecoverable;
+import static org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter.PendingFileRecoverable;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -44,22 +55,25 @@ public class FileWriterBucketStateSerializer
private final SimpleVersionedSerializer<InProgressFileRecoverable> inProgressFileRecoverableSerializer;
+ private final SimpleVersionedSerializer<PendingFileRecoverable> pendingFileRecoverableSerializer;
+
public FileWriterBucketStateSerializer(
- SimpleVersionedSerializer<InProgressFileRecoverable> inProgressFileRecoverableSerializer) {
- this.inProgressFileRecoverableSerializer = checkNotNull(
- inProgressFileRecoverableSerializer);
+ SimpleVersionedSerializer<InProgressFileRecoverable> inProgressFileRecoverableSerializer,
+ SimpleVersionedSerializer<PendingFileRecoverable> pendingFileRecoverableSerializer) {
+ this.inProgressFileRecoverableSerializer = checkNotNull(inProgressFileRecoverableSerializer);
+ this.pendingFileRecoverableSerializer = checkNotNull(pendingFileRecoverableSerializer);
}
@Override
public int getVersion() {
- return 1;
+ return 3;
}
@Override
public byte[] serialize(FileWriterBucketState state) throws IOException {
DataOutputSerializer out = new DataOutputSerializer(256);
out.writeInt(MAGIC_NUMBER);
- serializeV1(state, out);
+ serializeV3(state, out);
return out.getCopyOfBuffer();
}
@@ -73,12 +87,18 @@ public class FileWriterBucketStateSerializer
case 1:
validateMagicNumber(in);
return deserializeV1(in);
+ case 2:
+ validateMagicNumber(in);
+ return deserializeV2(in);
+ case 3:
+ validateMagicNumber(in);
+ return deserializeV3(in);
default:
throw new IOException("Unrecognized version or corrupt state: " + version);
}
}
- private void serializeV1(
+ private void serializeV3(
FileWriterBucketState state,
DataOutputView dataOutputView) throws IOException {
SimpleVersionedSerialization.writeVersionAndSerialize(
@@ -102,7 +122,41 @@ public class FileWriterBucketStateSerializer
}
}
- private FileWriterBucketState deserializeV1(DataInputView dataInputView) throws IOException {
+ private FileWriterBucketState deserializeV1(DataInputView in) throws IOException {
+ final SimpleVersionedSerializer<RecoverableWriter.CommitRecoverable> commitableSerializer = getCommitableSerializer();
+ final SimpleVersionedSerializer<RecoverableWriter.ResumeRecoverable> resumableSerializer = getResumableSerializer();
+
+ return internalDeserialize(
+ in,
+ dataInputView -> new OutputStreamBasedPartFileWriter.OutputStreamBasedInProgressFileRecoverable(
+ SimpleVersionedSerialization.readVersionAndDeSerialize(resumableSerializer, dataInputView)),
+ (version, bytes) -> new OutputStreamBasedPartFileWriter.OutputStreamBasedPendingFileRecoverable(
+ commitableSerializer.deserialize(version, bytes)));
+ }
+
+ private FileWriterBucketState deserializeV2(DataInputView in) throws IOException {
+ return internalDeserialize(
+ in,
+ dataInputView -> SimpleVersionedSerialization.readVersionAndDeSerialize(
+ inProgressFileRecoverableSerializer,
+ dataInputView),
+ pendingFileRecoverableSerializer::deserialize);
+ }
+
+ private FileWriterBucketState deserializeV3(DataInputView in) throws IOException {
+ return internalDeserialize(
+ in,
+ dataInputView -> SimpleVersionedSerialization.readVersionAndDeSerialize(
+ inProgressFileRecoverableSerializer,
+ dataInputView),
+ null);
+ }
+
+ private FileWriterBucketState internalDeserialize(
+ DataInputView dataInputView,
+ FunctionWithException<DataInputView, InProgressFileRecoverable, IOException> inProgressFileParser,
+ @Nullable BiFunctionWithException<Integer, byte[], PendingFileRecoverable, IOException> pendingFileParser) throws IOException {
+
String bucketId = SimpleVersionedSerialization.readVersionAndDeSerialize(
SimpleVersionedStringSerializer.INSTANCE,
dataInputView);
@@ -112,19 +166,37 @@ public class FileWriterBucketStateSerializer
// then get the current resumable stream
InProgressFileRecoverable current = null;
if (dataInputView.readBoolean()) {
- current = SimpleVersionedSerialization.readVersionAndDeSerialize(
- inProgressFileRecoverableSerializer,
- dataInputView);
+ current = inProgressFileParser.apply(dataInputView);
+ }
+
+ HashMap<Long, List<InProgressFileWriter.PendingFileRecoverable>> pendingFileRecoverablesPerCheckpoint = new HashMap<>();
+ if (pendingFileParser != null) {
+ final int pendingFileRecoverableSerializerVersion = dataInputView.readInt();
+ final int numCheckpoints = dataInputView.readInt();
+
+ for (int i = 0; i < numCheckpoints; i++) {
+ final long checkpointId = dataInputView.readLong();
+ final int numOfPendingFileRecoverables = dataInputView.readInt();
+
+ final List<InProgressFileWriter.PendingFileRecoverable> pendingFileRecoverables = new ArrayList<>(numOfPendingFileRecoverables);
+ for (int j = 0; j < numOfPendingFileRecoverables; j++) {
+ final byte[] bytes = new byte[dataInputView.readInt()];
+ dataInputView.readFully(bytes);
+ pendingFileRecoverables.add(pendingFileParser.apply(pendingFileRecoverableSerializerVersion, bytes));
+ }
+ pendingFileRecoverablesPerCheckpoint.put(checkpointId, pendingFileRecoverables);
+ }
}
return new FileWriterBucketState(
bucketId,
new Path(bucketPathStr),
creationTime,
- current);
+ current,
+ pendingFileRecoverablesPerCheckpoint);
}
- private static void validateMagicNumber(DataInputView in) throws IOException {
+ private void validateMagicNumber(DataInputView in) throws IOException {
int magicNumber = in.readInt();
if (magicNumber != MAGIC_NUMBER) {
throw new IOException(String.format(
@@ -132,4 +204,18 @@ public class FileWriterBucketStateSerializer
magicNumber));
}
}
+
+ private SimpleVersionedSerializer<RecoverableWriter.ResumeRecoverable> getResumableSerializer() {
+ final OutputStreamBasedPartFileWriter.OutputStreamBasedInProgressFileRecoverableSerializer
+ outputStreamBasedInProgressFileRecoverableSerializer =
+ (OutputStreamBasedPartFileWriter.OutputStreamBasedInProgressFileRecoverableSerializer) inProgressFileRecoverableSerializer;
+ return outputStreamBasedInProgressFileRecoverableSerializer.getResumeSerializer();
+ }
+
+ private SimpleVersionedSerializer<RecoverableWriter.CommitRecoverable> getCommitableSerializer() {
+ final OutputStreamBasedPartFileWriter.OutputStreamBasedPendingFileRecoverableSerializer
+ outputStreamBasedPendingFileRecoverableSerializer =
+ (OutputStreamBasedPartFileWriter.OutputStreamBasedPendingFileRecoverableSerializer) pendingFileRecoverableSerializer;
+ return outputStreamBasedPendingFileRecoverableSerializer.getCommitSerializer();
+ }
}
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerTest.java
index 445d00b6..d173434 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerTest.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerTest.java
@@ -70,7 +70,9 @@ public class FileWriterBucketStateSerializerTest {
private FileWriterBucketState serializeAndDeserialize(FileWriterBucketState bucketState) throws IOException {
FileWriterBucketStateSerializer serializer = new FileWriterBucketStateSerializer(
new FileSinkTestUtils.SimpleVersionedWrapperSerializer<>(
- FileSinkTestUtils.TestInProgressFileRecoverable::new));
+ FileSinkTestUtils.TestInProgressFileRecoverable::new),
+ new FileSinkTestUtils.SimpleVersionedWrapperSerializer<>(
+ FileSinkTestUtils.TestPendingFileRecoverable::new));
byte[] data = serializer.serialize(bucketState);
return serializer.deserialize(serializer.getVersion(), data);
}
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketTest.java
index 4d3f805..ec91dd7 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketTest.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketTest.java
@@ -31,6 +31,7 @@ import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
import org.apache.flink.core.fs.RecoverableWriter;
import org.apache.flink.core.fs.local.LocalFileSystem;
import org.apache.flink.core.fs.local.LocalRecoverableWriter;
+import org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputStreamBasedPartFileWriter;
import org.apache.flink.streaming.api.functions.sink.filesystem.PartFileInfo;
@@ -45,7 +46,10 @@ import org.junit.rules.TemporaryFolder;
import java.io.File;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.junit.Assert.assertEquals;
@@ -254,6 +258,28 @@ public class FileWriterBucketTest {
bucketState.getInProgressFileRecoverable());
}
+ /**
+ * Tests restoring with state containing pending files. This might happen
+ * if we are migrating from {@code StreamingFileSink}.
+ */
+ @Test
+ public void testRestoringWithOnlyPendingFiles() throws IOException {
+ final int noOfPendingFileCheckpoints = 4;
+
+ StubResumableWriter resumableWriter = new StubResumableWriter();
+ FileWriterBucket<String> bucket = getRestoredBucketWithOnlyPendingFiles(
+ resumableWriter,
+ DEFAULT_ROLLING_POLICY,
+ noOfPendingFileCheckpoints);
+ assertNull("There should be no in-progress file", bucket.getInProgressPart());
+ // There is one pending file for each checkpoint
+ assertEquals(noOfPendingFileCheckpoints, bucket.getPendingFiles().size());
+
+ List<FileSinkCommittable> fileSinkCommittables = bucket.prepareCommit(false);
+ bucket.snapshotState();
+ compareNumberOfPendingAndInProgress(fileSinkCommittables, noOfPendingFileCheckpoints, 0);
+ }
+
@Test
public void testMergeWithInprogressFileNotSupportResume() throws IOException {
FileWriterBucket<String> bucket1 = getRestoredBucketWithOnlyInProgressPart(
@@ -487,4 +513,37 @@ public class FileWriterBucketTest {
stateWithOnlyInProgressFile,
OutputFileConfig.builder().build());
}
+
+ private FileWriterBucket<String> getRestoredBucketWithOnlyPendingFiles(
+ BaseStubWriter writer,
+ RollingPolicy<String, String> rollingPolicy,
+ int numberOfPendingParts) throws IOException {
+
+ Map<Long, List<InProgressFileWriter.PendingFileRecoverable>> completePartsPerCheckpoint =
+ createPendingPartsPerCheckpoint(numberOfPendingParts);
+
+ FileWriterBucketState state =
+ new FileWriterBucketState(
+ "test",
+ new Path("file:///fake/fakefile"),
+ 12345L,
+ null,
+ completePartsPerCheckpoint);
+
+ return FileWriterBucket.restore(
+ new RowWiseBucketWriter<>(writer, ENCODER),
+ rollingPolicy,
+ state,
+ OutputFileConfig.builder().build());
+ }
+
+ private Map<Long, List<InProgressFileWriter.PendingFileRecoverable>> createPendingPartsPerCheckpoint(int noOfCheckpoints) {
+ final Map<Long, List<InProgressFileWriter.PendingFileRecoverable>> pendingCommittablesPerCheckpoint = new HashMap<>();
+ for (int checkpointId = 0; checkpointId < noOfCheckpoints; checkpointId++) {
+ final List<InProgressFileWriter.PendingFileRecoverable> pending = new ArrayList<>();
+ pending.add(new OutputStreamBasedPartFileWriter.OutputStreamBasedPendingFileRecoverable(new NoOpRecoverable()));
+ pendingCommittablesPerCheckpoint.put((long) checkpointId, pending);
+ }
+ return pendingCommittablesPerCheckpoint;
+ }
}
diff --git a/flink-connectors/flink-file-sink-common/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/OutputStreamBasedPartFileWriter.java b/flink-connectors/flink-file-sink-common/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/OutputStreamBasedPartFileWriter.java
index 64c2bae..715bf33 100644
--- a/flink-connectors/flink-file-sink-common/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/OutputStreamBasedPartFileWriter.java
+++ b/flink-connectors/flink-file-sink-common/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/OutputStreamBasedPartFileWriter.java
@@ -135,11 +135,14 @@ public abstract class OutputStreamBasedPartFileWriter<IN, BucketID> extends Abst
final long creationTime) throws IOException;
}
- static final class OutputStreamBasedPendingFileRecoverable implements PendingFileRecoverable {
+ /**
+ * The {@link PendingFileRecoverable} implementation for {@link OutputStreamBasedBucketWriter}.
+ */
+ public static final class OutputStreamBasedPendingFileRecoverable implements PendingFileRecoverable {
private final RecoverableWriter.CommitRecoverable commitRecoverable;
- OutputStreamBasedPendingFileRecoverable(final RecoverableWriter.CommitRecoverable commitRecoverable) {
+ public OutputStreamBasedPendingFileRecoverable(final RecoverableWriter.CommitRecoverable commitRecoverable) {
this.commitRecoverable = commitRecoverable;
}
@@ -149,7 +152,7 @@ public abstract class OutputStreamBasedPartFileWriter<IN, BucketID> extends Abst
}
/**
- * A kind of {@link InProgressFileRecoverable} implementation.
+ * The {@link InProgressFileRecoverable} implementation for {@link OutputStreamBasedBucketWriter}.
*/
public static final class OutputStreamBasedInProgressFileRecoverable implements InProgressFileRecoverable {
@@ -183,7 +186,10 @@ public abstract class OutputStreamBasedPartFileWriter<IN, BucketID> extends Abst
}
}
- static class OutputStreamBasedInProgressFileRecoverableSerializer implements SimpleVersionedSerializer<InProgressFileRecoverable> {
+ /**
+ * The serializer for {@link OutputStreamBasedInProgressFileRecoverable}.
+ */
+ public static class OutputStreamBasedInProgressFileRecoverableSerializer implements SimpleVersionedSerializer<InProgressFileRecoverable> {
private static final int MAGIC_NUMBER = 0xb3a4073d;
@@ -219,7 +225,7 @@ public abstract class OutputStreamBasedPartFileWriter<IN, BucketID> extends Abst
}
}
- SimpleVersionedSerializer<RecoverableWriter.ResumeRecoverable> getResumeSerializer() {
+ public SimpleVersionedSerializer<RecoverableWriter.ResumeRecoverable> getResumeSerializer() {
return resumeSerializer;
}
@@ -239,7 +245,10 @@ public abstract class OutputStreamBasedPartFileWriter<IN, BucketID> extends Abst
}
}
- static class OutputStreamBasedPendingFileRecoverableSerializer implements SimpleVersionedSerializer<PendingFileRecoverable> {
+ /**
+ * The serializer for {@link OutputStreamBasedPendingFileRecoverable}.
+ */
+ public static class OutputStreamBasedPendingFileRecoverableSerializer implements SimpleVersionedSerializer<PendingFileRecoverable> {
private static final int MAGIC_NUMBER = 0x2c853c89;
@@ -276,7 +285,7 @@ public abstract class OutputStreamBasedPartFileWriter<IN, BucketID> extends Abst
}
}
- SimpleVersionedSerializer<RecoverableWriter.CommitRecoverable> getCommitSerializer() {
+ public SimpleVersionedSerializer<RecoverableWriter.CommitRecoverable> getCommitSerializer() {
return this.commitSerializer;
}
[flink] 04/10: [FLINK-20337] Extend StatefulSinkWriterOperator
Javadoc
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 073c9526a545baa727beef04c776ab12e82cbe24
Author: Aljoscha Krettek <al...@apache.org>
AuthorDate: Fri Nov 27 12:41:17 2020 +0100
[FLINK-20337] Extend StatefulSinkWriterOperator Javadoc
---
.../operators/sink/StatefulSinkWriterOperator.java | 20 ++++++++++++++++----
1 file changed, 16 insertions(+), 4 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperator.java
index 65b395c..769dc29 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterOperator.java
@@ -59,13 +59,21 @@ final class StatefulSinkWriterOperator<InputT, CommT, WriterStateT> extends Abst
/** The writer operator's state serializer. */
private final SimpleVersionedSerializer<WriterStateT> writerStateSimpleVersionedSerializer;
- /** The previous sink operator's state name. */
+ /**
+ * The previous sink operator's state name. We allow restoring state from a different
+ * (compatible) sink implementation such as {@link org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink}.
+ * This allows migration to newer Sink implementations.
+ */
@Nullable
private final String previousSinkStateName;
// ------------------------------- runtime fields ---------------------------------------
- /** The previous sink operator's state. */
+ /**
+ * The previous sink operator's state. We allow restoring state from a different (compatible)
+ * sink implementation such as {@link org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink}.
+ * This allows migration to newer Sink implementations.
+ */
@Nullable
private ListState<WriterStateT> previousSinkState;
@@ -87,8 +95,12 @@ final class StatefulSinkWriterOperator<InputT, CommT, WriterStateT> extends Abst
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
- final ListState<byte[]> rawState = context.getOperatorStateStore().getListState(WRITER_RAW_STATES_DESC);
- writerState = new SimpleVersionedListState<>(rawState, writerStateSimpleVersionedSerializer);
+ final ListState<byte[]> rawState = context
+ .getOperatorStateStore()
+ .getListState(WRITER_RAW_STATES_DESC);
+ writerState = new SimpleVersionedListState<>(
+ rawState,
+ writerStateSimpleVersionedSerializer);
if (previousSinkStateName != null) {
final ListStateDescriptor<byte[]> preSinkStateDesc = new ListStateDescriptor<>(
[flink] 07/10: [FLINK-20337] Add migration test for deserializing
existing StreamingFileSink state with FileSink serializer
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 63f7f3b311a5fe8496014c54243c576e37559b68
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Fri Nov 27 19:40:27 2020 +0800
[FLINK-20337] Add migration test for deserializing existing StreamingFileSink state with FileSink serializer
This re-uses the generated test data from BucketStateSerializerTest.
---
.../flink-connector-migration-test/pom.xml | 58 +++++
...leWriterBucketStateSerializerMigrationTest.java | 285 +++++++++++++++++++++
.../bucket-state-migration-test/empty-v1/snapshot | Bin 0 -> 128 bytes
.../bucket-state-migration-test/empty-v2/snapshot | Bin 0 -> 128 bytes
...inprogress.a88d5993-77bc-44ce-880b-9f2a43b59ab4 | 2 +
...inprogress.7c0f2bd7-3078-48e8-9af2-d8773fb949c5 | 2 +
...inprogress.6729a640-0585-4785-a652-89802950c663 | 2 +
...inprogress.b4bcb0e9-5c9e-45dd-8963-1b163343544d | 2 +
...inprogress.e1e9e48d-0db6-4dd7-8a4d-fb4ebe7ed8ac | 2 +
.../full-no-in-progress-v1-template/snapshot | Bin 0 -> 1537 bytes
...inprogress.1e22e72d-0ab2-493b-8b00-9edac4252cec | 2 +
...inprogress.3821f491-9fa1-48b2-b66b-655352a3c8ec | 2 +
...inprogress.0af18f41-d8f8-4a4e-a92e-de12851be20b | 2 +
...inprogress.a3d0f4d2-d6ad-4f83-ba62-ed4b1fa86db2 | 2 +
...inprogress.666acf3e-935c-4621-8171-f7c897496524 | 2 +
.../full-no-in-progress-v2-template/snapshot | Bin 0 -> 1597 bytes
...inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82 | 2 +
...inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c | 2 +
...inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97 | 2 +
...inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9 | 2 +
...inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7 | 2 +
...inprogress.32f5a28f-20e1-48da-9951-10e795133d64 | 1 +
.../full-v1-template/snapshot | Bin 0 -> 1613 bytes
...inprogress.9731063e-2b28-4701-8cc1-e706480b8022 | 2 +
...inprogress.1d423406-097a-4deb-bfde-d023d3477cd5 | 2 +
...inprogress.6a837aa3-4736-4098-a878-fdeffe227628 | 2 +
...inprogress.f121b73d-ac74-4fbd-b70d-f13e51c9132c | 2 +
...inprogress.a156884a-f090-4c3f-a271-0b63ab539c45 | 2 +
...inprogress.83c527c5-14dc-4d49-9f99-c915f2224f6a | 1 +
.../full-v2-template/snapshot | Bin 0 -> 1685 bytes
...inprogress.a70190d6-d080-43a8-b414-746b09d3a8a0 | 1 +
.../only-in-progress-v1/snapshot | Bin 0 -> 404 bytes
...inprogress.10833090-dd8c-4e36-884d-bb9758a3a8ef | 1 +
.../only-in-progress-v2/snapshot | Bin 0 -> 416 bytes
.../src/test/resources/log4j2-test.properties | 29 +++
flink-connectors/pom.xml | 1 +
36 files changed, 417 insertions(+)
diff --git a/flink-connectors/flink-connector-migration-test/pom.xml b/flink-connectors/flink-connector-migration-test/pom.xml
new file mode 100644
index 0000000..9a16ada
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/pom.xml
@@ -0,0 +1,58 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connectors</artifactId>
+ <version>1.12-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-file-migration-test</artifactId>
+ <name>Flink : Connectors : Migration Test</name>
+
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-files</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerMigrationTest.java b/flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerMigrationTest.java
new file mode 100644
index 0000000..4cce402
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/java/org/apache/flink/connector/file/sink/writer/FileWriterBucketStateSerializerMigrationTest.java
@@ -0,0 +1,285 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.sink.writer;
+
+import org.apache.flink.api.common.serialization.SimpleStringEncoder;
+import org.apache.flink.connector.file.sink.FileSinkCommittable;
+import org.apache.flink.connector.file.sink.committer.FileCommitter;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.io.SimpleVersionedSerialization;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.functions.sink.filesystem.BucketStateGenerator;
+import org.apache.flink.streaming.api.functions.sink.filesystem.BucketStatePathResolver;
+import org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter;
+import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
+import org.apache.flink.streaming.api.functions.sink.filesystem.RowWiseBucketWriter;
+import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
+import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
+import org.apache.flink.util.FileUtils;
+
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.startsWith;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.iterableWithSize;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for the {@link FileWriterBucketStateSerializer} that verify we can still read snapshots
+ * taken from old {@link StreamingFileSink}. We keep snapshots for all previous versions in version
+ * control (including the current version). The tests verify that the current version of the
+ * serializer can still read data from all previous versions.
+ *
+ * <p>This is a mirror of {@link org.apache.flink.streaming.api.functions.sink.filesystem.BucketStateSerializerTest}
+ * that verifies we can restore the same snapshots. We therefore have the same "previous versions"
+ * as that other test. The generated test data from {@code BucketStateSerializerTest} has been
+ * copied to be reused in this test to ensure we can restore the same bytes.
+ */
+@RunWith(Parameterized.class)
+public class FileWriterBucketStateSerializerMigrationTest {
+
+ private static final int CURRENT_VERSION = 2;
+
+ @Parameterized.Parameters(name = "Previous Version = {0}")
+ public static Collection<Integer> previousVersions() {
+ return Arrays.asList(1, 2);
+ }
+
+ @Parameterized.Parameter
+ public Integer previousVersion;
+
+ private static final String IN_PROGRESS_CONTENT = "writing";
+ private static final String PENDING_CONTENT = "wrote";
+
+ private static final String BUCKET_ID = "test-bucket";
+
+ private static final java.nio.file.Path BASE_PATH = Paths.get("src/test/resources/")
+ .resolve("bucket-state-migration-test");
+
+ @ClassRule
+ public static TemporaryFolder tempFolder = new TemporaryFolder();
+
+ private final BucketStateGenerator generator = new BucketStateGenerator(
+ BUCKET_ID,
+ IN_PROGRESS_CONTENT,
+ PENDING_CONTENT,
+ BASE_PATH,
+ CURRENT_VERSION);
+
+ @Test
+ @Ignore
+ public void prepareDeserializationEmpty() throws IOException {
+ generator.prepareDeserializationEmpty();
+ }
+
+ @Test
+ public void testSerializationEmpty() throws IOException {
+
+ final String scenarioName = "empty";
+ final BucketStatePathResolver pathResolver = new BucketStatePathResolver(BASE_PATH, previousVersion);
+
+ final java.nio.file.Path outputPath = pathResolver.getOutputPath(scenarioName);
+ final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
+ final FileWriterBucketState recoveredState = readBucketState(scenarioName, previousVersion);
+
+ final FileWriterBucket<String> bucket = restoreBucket(recoveredState);
+
+ Assert.assertEquals(testBucketPath, bucket.getBucketPath());
+ Assert.assertNull(bucket.getInProgressPart());
+ Assert.assertTrue(bucket.getPendingFiles().isEmpty());
+ }
+
+ @Test
+ @Ignore
+ public void prepareDeserializationOnlyInProgress() throws IOException {
+ generator.prepareDeserializationOnlyInProgress();
+ }
+
+ @Test
+ public void testSerializationOnlyInProgress() throws IOException {
+
+ final String scenarioName = "only-in-progress";
+ final BucketStatePathResolver pathResolver = new BucketStatePathResolver(BASE_PATH, previousVersion);
+
+ final java.nio.file.Path outputPath = pathResolver.getOutputPath(scenarioName);
+
+ final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
+
+ final FileWriterBucketState recoveredState = readBucketState(scenarioName, previousVersion);
+
+ final FileWriterBucket<String> bucket = restoreBucket(recoveredState);
+
+ Assert.assertEquals(testBucketPath, bucket.getBucketPath());
+
+ //check restore the correct in progress file writer
+ Assert.assertEquals(8, bucket.getInProgressPart().getSize());
+
+ long numFiles = Files.list(Paths.get(testBucketPath.toString()))
+ .map(file -> {
+ assertThat(
+ file.getFileName().toString(),
+ startsWith(".part-0-0.inprogress"));
+ return 1;
+ })
+ .count();
+
+ assertThat(numFiles, is(1L));
+ }
+
+ @Test
+ @Ignore
+ public void prepareDeserializationFull() throws IOException {
+ generator.prepareDeserializationFull();
+ }
+
+ @Test
+ public void testSerializationFull() throws IOException {
+ testDeserializationFull(true, "full");
+ }
+
+ @Test
+ @Ignore
+ public void prepareDeserializationNullInProgress() throws IOException {
+ generator.prepareDeserializationNullInProgress();
+ }
+
+ @Test
+ public void testSerializationNullInProgress() throws IOException {
+ testDeserializationFull(false, "full-no-in-progress");
+ }
+
+ private void testDeserializationFull(final boolean withInProgress, final String scenarioName) throws IOException {
+
+ final BucketStatePathResolver pathResolver = new BucketStatePathResolver(BASE_PATH, previousVersion);
+
+ try {
+ final java.nio.file.Path outputPath = pathResolver.getOutputPath(scenarioName);
+ final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
+ // restore the state
+ final FileWriterBucketState recoveredState = readBucketStateFromTemplate(scenarioName, previousVersion);
+ final int noOfPendingCheckpoints = 5;
+
+ // there are 5 checkpoint does not complete.
+ final Map<Long, List<InProgressFileWriter.PendingFileRecoverable>>
+ pendingFileRecoverables = recoveredState.getPendingFileRecoverablesPerCheckpoint();
+ Assert.assertEquals(5L, pendingFileRecoverables.size());
+
+ final Set<String> beforeRestorePaths = Files.list(outputPath.resolve(BUCKET_ID))
+ .map(file -> file.getFileName().toString())
+ .collect(Collectors.toSet());
+
+ // before retsoring all file has "inprogress"
+ for (int i = 0; i < noOfPendingCheckpoints; i++) {
+ final String part = ".part-0-" + i + ".inprogress";
+ assertThat(beforeRestorePaths, hasItem(startsWith(part)));
+ }
+
+ final FileWriterBucket<String> bucket = restoreBucket(recoveredState);
+ Assert.assertEquals(testBucketPath, bucket.getBucketPath());
+ Assert.assertEquals(noOfPendingCheckpoints, bucket.getPendingFiles().size());
+
+ // simulates we commit the recovered pending files on the first checkpoint
+ bucket.snapshotState();
+ List<FileSinkCommittable> committables = bucket.prepareCommit(false);
+ FileCommitter committer = new FileCommitter(createBucketWriter());
+ committer.commit(committables);
+
+ final Set<String> afterRestorePaths = Files.list(outputPath.resolve(BUCKET_ID))
+ .map(file -> file.getFileName().toString())
+ .collect(Collectors.toSet());
+
+ // after restoring all pending files are comitted.
+ // there is no "inporgress" in file name for the committed files.
+ for (int i = 0; i < noOfPendingCheckpoints; i++) {
+ final String part = "part-0-" + i;
+ assertThat(afterRestorePaths, hasItem(part));
+ afterRestorePaths.remove(part);
+ }
+
+ if (withInProgress) {
+ // only the in-progress must be left
+ assertThat(afterRestorePaths, iterableWithSize(1));
+
+ // verify that the in-progress file is still there
+ assertThat(afterRestorePaths, hasItem(startsWith(".part-0-" + noOfPendingCheckpoints + ".inprogress")));
+ } else {
+ assertThat(afterRestorePaths, empty());
+ }
+ } finally {
+ FileUtils.deleteDirectory(pathResolver.getResourcePath(scenarioName).toFile());
+ }
+ }
+
+ private static FileWriterBucket<String> restoreBucket(final FileWriterBucketState bucketState) throws IOException {
+ return FileWriterBucket.restore(
+ createBucketWriter(),
+ DefaultRollingPolicy.builder().withMaxPartSize(10).build(),
+ bucketState,
+ OutputFileConfig.builder().build());
+ }
+
+ private static RowWiseBucketWriter<String, String> createBucketWriter() throws IOException {
+ return new RowWiseBucketWriter<>(FileSystem.getLocalFileSystem().createRecoverableWriter(), new SimpleStringEncoder<>());
+ }
+
+ private static SimpleVersionedSerializer<FileWriterBucketState> bucketStateSerializer() throws IOException {
+ final RowWiseBucketWriter<String, String> bucketWriter = createBucketWriter();
+ return new FileWriterBucketStateSerializer(
+ bucketWriter.getProperties().getInProgressFileRecoverableSerializer(),
+ bucketWriter.getProperties().getPendingFileRecoverableSerializer());
+ }
+
+ private static FileWriterBucketState readBucketState(final String scenarioName, final int version) throws IOException {
+ final BucketStatePathResolver pathResolver = new BucketStatePathResolver(BASE_PATH, version);
+ byte[] bytes = Files.readAllBytes(pathResolver.getSnapshotPath(scenarioName));
+ return SimpleVersionedSerialization.readVersionAndDeSerialize(bucketStateSerializer(), bytes);
+ }
+
+ private static FileWriterBucketState readBucketStateFromTemplate(final String scenarioName, final int version) throws IOException {
+ final BucketStatePathResolver pathResolver = new BucketStatePathResolver(BASE_PATH, version);
+ final java.nio.file.Path scenarioPath = pathResolver.getResourcePath(scenarioName);
+
+ // clear the scenario files first
+ FileUtils.deleteDirectory(scenarioPath.toFile());
+
+ // prepare the scenario files
+ FileUtils.copy(new Path(scenarioPath.toString() + "-template"), new Path(scenarioPath.toString()), false);
+
+ return readBucketState(scenarioName, version);
+ }
+}
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/empty-v1/snapshot b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/empty-v1/snapshot
new file mode 100644
index 0000000..9700f3d
Binary files /dev/null and b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/empty-v1/snapshot differ
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/empty-v2/snapshot b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/empty-v2/snapshot
new file mode 100644
index 0000000..9e84e8d
Binary files /dev/null and b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/empty-v2/snapshot differ
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-0.inprogress.a88d5993-77bc-44ce-880b-9f2a43b59ab4 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-0.inprogress.a88d5993-77bc-44ce-880b-9f2a43b59ab4
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-0.inprogress.a88d5993-77bc-44ce-880b-9f2a43b59ab4
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-1.inprogress.7c0f2bd7-3078-48e8-9af2-d8773fb949c5 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-1.inprogress.7c0f2bd7-3078-48e8-9af2-d8773fb949c5
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-1.inprogress.7c0f2bd7-3078-48e8-9af2-d8773fb949c5
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-2.inprogress.6729a640-0585-4785-a652-89802950c663 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-2.inprogress.6729a640-0585-4785-a652-89802950c663
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-2.inprogress.6729a640-0585-4785-a652-89802950c663
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-3.inprogress.b4bcb0e9-5c9e-45dd-8963-1b163343544d b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-3.inprogress.b4bcb0e9-5c9e-45dd-8963-1b163343544d
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-3.inprogress.b4bcb0e9-5c9e-45dd-8963-1b163343544d
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-4.inprogress.e1e9e48d-0db6-4dd7-8a4d-fb4ebe7ed8ac b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-4.inprogress.e1e9e48d-0db6-4dd7-8a4d-fb4ebe7ed8ac
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/bucket/test-bucket/.part-0-4.inprogress.e1e9e48d-0db6-4dd7-8a4d-fb4ebe7ed8ac
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/snapshot b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/snapshot
new file mode 100644
index 0000000..5d6644c
Binary files /dev/null and b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/snapshot differ
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-0.inprogress.1e22e72d-0ab2-493b-8b00-9edac4252cec b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-0.inprogress.1e22e72d-0ab2-493b-8b00-9edac4252cec
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-0.inprogress.1e22e72d-0ab2-493b-8b00-9edac4252cec
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-1.inprogress.3821f491-9fa1-48b2-b66b-655352a3c8ec b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-1.inprogress.3821f491-9fa1-48b2-b66b-655352a3c8ec
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-1.inprogress.3821f491-9fa1-48b2-b66b-655352a3c8ec
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-2.inprogress.0af18f41-d8f8-4a4e-a92e-de12851be20b b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-2.inprogress.0af18f41-d8f8-4a4e-a92e-de12851be20b
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-2.inprogress.0af18f41-d8f8-4a4e-a92e-de12851be20b
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-3.inprogress.a3d0f4d2-d6ad-4f83-ba62-ed4b1fa86db2 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-3.inprogress.a3d0f4d2-d6ad-4f83-ba62-ed4b1fa86db2
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-3.inprogress.a3d0f4d2-d6ad-4f83-ba62-ed4b1fa86db2
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-4.inprogress.666acf3e-935c-4621-8171-f7c897496524 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-4.inprogress.666acf3e-935c-4621-8171-f7c897496524
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/bucket/test-bucket/.part-0-4.inprogress.666acf3e-935c-4621-8171-f7c897496524
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/snapshot b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/snapshot
new file mode 100644
index 0000000..4e0eaa4
Binary files /dev/null and b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-no-in-progress-v2-template/snapshot differ
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-0.inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-0.inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-0.inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-1.inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-1.inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-1.inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-2.inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-2.inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-2.inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-3.inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-3.inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-3.inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-4.inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-4.inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-4.inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-5.inprogress.32f5a28f-20e1-48da-9951-10e795133d64 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-5.inprogress.32f5a28f-20e1-48da-9951-10e795133d64
new file mode 100644
index 0000000..631ee76
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-5.inprogress.32f5a28f-20e1-48da-9951-10e795133d64
@@ -0,0 +1 @@
+writing
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/snapshot b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/snapshot
new file mode 100644
index 0000000..dd313db
Binary files /dev/null and b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v1-template/snapshot differ
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-0.inprogress.9731063e-2b28-4701-8cc1-e706480b8022 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-0.inprogress.9731063e-2b28-4701-8cc1-e706480b8022
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-0.inprogress.9731063e-2b28-4701-8cc1-e706480b8022
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-1.inprogress.1d423406-097a-4deb-bfde-d023d3477cd5 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-1.inprogress.1d423406-097a-4deb-bfde-d023d3477cd5
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-1.inprogress.1d423406-097a-4deb-bfde-d023d3477cd5
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-2.inprogress.6a837aa3-4736-4098-a878-fdeffe227628 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-2.inprogress.6a837aa3-4736-4098-a878-fdeffe227628
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-2.inprogress.6a837aa3-4736-4098-a878-fdeffe227628
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-3.inprogress.f121b73d-ac74-4fbd-b70d-f13e51c9132c b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-3.inprogress.f121b73d-ac74-4fbd-b70d-f13e51c9132c
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-3.inprogress.f121b73d-ac74-4fbd-b70d-f13e51c9132c
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-4.inprogress.a156884a-f090-4c3f-a271-0b63ab539c45 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-4.inprogress.a156884a-f090-4c3f-a271-0b63ab539c45
new file mode 100644
index 0000000..a6c0130
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-4.inprogress.a156884a-f090-4c3f-a271-0b63ab539c45
@@ -0,0 +1,2 @@
+wrote
+wrote
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-5.inprogress.83c527c5-14dc-4d49-9f99-c915f2224f6a b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-5.inprogress.83c527c5-14dc-4d49-9f99-c915f2224f6a
new file mode 100644
index 0000000..631ee76
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/bucket/test-bucket/.part-0-5.inprogress.83c527c5-14dc-4d49-9f99-c915f2224f6a
@@ -0,0 +1 @@
+writing
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/snapshot b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/snapshot
new file mode 100644
index 0000000..9d9c3fc
Binary files /dev/null and b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/full-v2-template/snapshot differ
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v1/bucket/test-bucket/.part-0-0.inprogress.a70190d6-d080-43a8-b414-746b09d3a8a0 b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v1/bucket/test-bucket/.part-0-0.inprogress.a70190d6-d080-43a8-b414-746b09d3a8a0
new file mode 100644
index 0000000..631ee76
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v1/bucket/test-bucket/.part-0-0.inprogress.a70190d6-d080-43a8-b414-746b09d3a8a0
@@ -0,0 +1 @@
+writing
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v1/snapshot b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v1/snapshot
new file mode 100644
index 0000000..6e98be9
Binary files /dev/null and b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v1/snapshot differ
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v2/bucket/test-bucket/.part-0-0.inprogress.10833090-dd8c-4e36-884d-bb9758a3a8ef b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v2/bucket/test-bucket/.part-0-0.inprogress.10833090-dd8c-4e36-884d-bb9758a3a8ef
new file mode 100644
index 0000000..631ee76
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v2/bucket/test-bucket/.part-0-0.inprogress.10833090-dd8c-4e36-884d-bb9758a3a8ef
@@ -0,0 +1 @@
+writing
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v2/snapshot b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v2/snapshot
new file mode 100644
index 0000000..d21f7c4
Binary files /dev/null and b/flink-connectors/flink-connector-migration-test/src/test/resources/bucket-state-migration-test/only-in-progress-v2/snapshot differ
diff --git a/flink-connectors/flink-connector-migration-test/src/test/resources/log4j2-test.properties b/flink-connectors/flink-connector-migration-test/src/test/resources/log4j2-test.properties
new file mode 100644
index 0000000..6faab06
--- /dev/null
+++ b/flink-connectors/flink-connector-migration-test/src/test/resources/log4j2-test.properties
@@ -0,0 +1,29 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
+
diff --git a/flink-connectors/pom.xml b/flink-connectors/pom.xml
index 11ef77e..1f91f6c 100644
--- a/flink-connectors/pom.xml
+++ b/flink-connectors/pom.xml
@@ -57,6 +57,7 @@ under the License.
<module>flink-connector-base</module>
<module>flink-file-sink-common</module>
<module>flink-connector-files</module>
+ <module>flink-connector-migration-test</module>
</modules>
<!-- override these root dependencies as 'provided', so they don't end up
[flink] 01/10: [hotfix][fs-connector] Remove unused state
serializer from FileWriter
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 99a8891fe841c32f9ec07d67a124f8eacd954af2
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Fri Nov 27 18:11:00 2020 +0800
[hotfix][fs-connector] Remove unused state serializer from FileWriter
---
.../org/apache/flink/connector/file/sink/writer/FileWriter.java | 9 +--------
1 file changed, 1 insertion(+), 8 deletions(-)
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriter.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriter.java
index 5c2faaa..aaf0473 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriter.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/writer/FileWriter.java
@@ -85,10 +85,6 @@ public class FileWriter<IN> implements
private final OutputFileConfig outputFileConfig;
- // --------------------------- State Related Fields -----------------------------
-
- private final FileWriterBucketStateSerializer bucketStateSerializer;
-
/**
* A constructor creating a new empty bucket manager.
*
@@ -119,9 +115,6 @@ public class FileWriter<IN> implements
this.activeBuckets = new HashMap<>();
this.bucketerContext = new BucketerContext();
- this.bucketStateSerializer = new FileWriterBucketStateSerializer(
- bucketWriter.getProperties().getInProgressFileRecoverableSerializer());
-
this.processingTimeService = checkNotNull(processingTimeService);
checkArgument(
bucketCheckInterval > 0,
@@ -216,7 +209,7 @@ public class FileWriter<IN> implements
@Override
public List<FileWriterBucketState> snapshotState() throws IOException {
checkState(
- bucketWriter != null && bucketStateSerializer != null,
+ bucketWriter != null,
"sink has not been initialized");
List<FileWriterBucketState> state = new ArrayList<>();
[flink] 02/10: [hotfix][fs-connector] Fix the typo in the
FileSinkITCase
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4bd4e76a2d55aafec960dae326ab0e326463620c
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Fri Nov 27 18:14:26 2020 +0800
[hotfix][fs-connector] Fix the typo in the FileSinkITCase
---
.../flink/connector/file/sink/StreamingExecutionFileSinkITCase.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/StreamingExecutionFileSinkITCase.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/StreamingExecutionFileSinkITCase.java
index 0c10291..4fefde1 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/StreamingExecutionFileSinkITCase.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/StreamingExecutionFileSinkITCase.java
@@ -92,7 +92,7 @@ public class StreamingExecutionFileSinkITCase extends FileSinkITBase {
}
env.addSource(new StreamingExecutionTestSource(latchId, NUM_RECORDS, triggerFailover))
- .setParallelism(NUM_BUCKETS)
+ .setParallelism(NUM_SOURCES)
.sinkTo(createFileSink(path))
.setParallelism(NUM_SINKS);
[flink] 08/10: [refactor] Factor common testing code out of
FileSinkITBase
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.12
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0528a21f7768c5c653bcbe03437c7ab9681dcf86
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Thu Nov 26 17:46:59 2020 +0800
[refactor] Factor common testing code out of FileSinkITBase
We want to reuse this code for the new FileSink migration test as well.
---
flink-connectors/flink-connector-files/pom.xml | 15 +++
.../flink/connector/file/sink/FileSinkITBase.java | 93 +-------------
.../sink/utils/IntegerFileSinkTestDataUtils.java | 140 +++++++++++++++++++++
3 files changed, 159 insertions(+), 89 deletions(-)
diff --git a/flink-connectors/flink-connector-files/pom.xml b/flink-connectors/flink-connector-files/pom.xml
index 117e2c5..65a9027 100644
--- a/flink-connectors/flink-connector-files/pom.xml
+++ b/flink-connectors/flink-connector-files/pom.xml
@@ -92,4 +92,19 @@ under the License.
</dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
</project>
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/FileSinkITBase.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/FileSinkITBase.java
index 6cd09d6..4912f2c 100644
--- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/FileSinkITBase.java
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/FileSinkITBase.java
@@ -18,17 +18,14 @@
package org.apache.flink.connector.file.sink;
-import org.apache.flink.api.common.serialization.Encoder;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.connector.file.sink.utils.IntegerFileSinkTestDataUtils;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
-import org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner;
import org.apache.flink.streaming.api.functions.sink.filesystem.PartFileInfo;
-import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.CheckpointRollingPolicy;
import org.apache.flink.util.TestLogger;
@@ -37,22 +34,9 @@ import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runners.Parameterized;
-import java.io.DataInputStream;
-import java.io.EOFException;
-import java.io.File;
-import java.io.FileInputStream;
import java.io.IOException;
-import java.io.OutputStream;
-import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
/**
* The base class for the File Sink IT Case in different execution mode.
@@ -101,88 +85,19 @@ public abstract class FileSinkITBase extends TestLogger {
miniCluster.executeJobBlocking(jobGraph);
}
- checkResult(path);
+ IntegerFileSinkTestDataUtils.checkIntegerSequenceSinkOutput(path, NUM_RECORDS, NUM_BUCKETS, NUM_SOURCES);
}
protected abstract JobGraph createJobGraph(String path);
protected FileSink<Integer> createFileSink(String path) {
return FileSink
- .forRowFormat(new Path(path), new IntEncoder())
- .withBucketAssigner(new ModuloBucketAssigner())
+ .forRowFormat(new Path(path), new IntegerFileSinkTestDataUtils.IntEncoder())
+ .withBucketAssigner(new IntegerFileSinkTestDataUtils.ModuloBucketAssigner(NUM_BUCKETS))
.withRollingPolicy(new PartSizeAndCheckpointRollingPolicy(1024))
.build();
}
- private void checkResult(String path) throws Exception {
- File dir = new File(path);
- String[] subDirNames = dir.list();
- assertNotNull(subDirNames);
-
- Arrays.sort(subDirNames, Comparator.comparingInt(Integer::parseInt));
- assertEquals(NUM_BUCKETS, subDirNames.length);
- for (int i = 0; i < NUM_BUCKETS; ++i) {
- assertEquals(Integer.toString(i), subDirNames[i]);
-
- // now check its content
- File bucketDir = new File(path, subDirNames[i]);
- assertTrue(
- bucketDir.getAbsolutePath() + " Should be a existing directory",
- bucketDir.isDirectory());
-
- Map<Integer, Integer> counts = new HashMap<>();
- File[] files = bucketDir.listFiles(f -> !f.getName().startsWith("."));
- assertNotNull(files);
-
- for (File file : files) {
- assertTrue(file.isFile());
-
- try (DataInputStream dataInputStream = new DataInputStream(new FileInputStream(file))) {
- while (true) {
- int value = dataInputStream.readInt();
- counts.compute(value, (k, v) -> v == null ? 1 : v + 1);
- }
- } catch (EOFException e) {
- // End the reading
- }
- }
-
- int expectedCount = NUM_RECORDS / NUM_BUCKETS +
- (i < NUM_RECORDS % NUM_BUCKETS ? 1 : 0);
- assertEquals(expectedCount, counts.size());
-
- for (int j = i; j < NUM_RECORDS; j += NUM_BUCKETS) {
- assertEquals(
- "The record " + j + " should occur " + NUM_SOURCES + " times, " +
- " but only occurs " + counts.getOrDefault(j, 0) + "time",
- NUM_SOURCES,
- counts.getOrDefault(j, 0).intValue());
- }
- }
- }
-
- private static class IntEncoder implements Encoder<Integer> {
-
- @Override
- public void encode(Integer element, OutputStream stream) throws IOException {
- stream.write(ByteBuffer.allocate(4).putInt(element).array());
- stream.flush();
- }
- }
-
- private static class ModuloBucketAssigner implements BucketAssigner<Integer, String> {
-
- @Override
- public String getBucketId(Integer element, Context context) {
- return Integer.toString(element % NUM_BUCKETS);
- }
-
- @Override
- public SimpleVersionedSerializer<String> getSerializer() {
- return SimpleVersionedStringSerializer.INSTANCE;
- }
- }
-
private static class PartSizeAndCheckpointRollingPolicy
extends CheckpointRollingPolicy<Integer, String> {
diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/utils/IntegerFileSinkTestDataUtils.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/utils/IntegerFileSinkTestDataUtils.java
new file mode 100644
index 0000000..a0dab46
--- /dev/null
+++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/sink/utils/IntegerFileSinkTestDataUtils.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.file.sink.utils;
+
+import org.apache.flink.api.common.serialization.Encoder;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner;
+import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
+
+import java.io.DataInputStream;
+import java.io.EOFException;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Utilities for file sinks that writes a sequence of continues integers into files starting from 0.
+ * The sinks expect multiple sources writing the same sequence onto the disk and the integers are
+ * assigned to different buckets according to modulo.
+ */
+public class IntegerFileSinkTestDataUtils {
+
+ /**
+ * Testing sink {@link Encoder} that writes integer with its binary representation.
+ */
+ public static class IntEncoder implements Encoder<Integer> {
+
+ @Override
+ public void encode(Integer element, OutputStream stream) throws IOException {
+ stream.write(ByteBuffer.allocate(4).putInt(element).array());
+ stream.flush();
+ }
+ }
+
+ /**
+ * Testing {@link BucketAssigner} that assigns integers according to modulo.
+ */
+ public static class ModuloBucketAssigner implements BucketAssigner<Integer, String> {
+
+ private final int numBuckets;
+
+ public ModuloBucketAssigner(int numBuckets) {
+ this.numBuckets = numBuckets;
+ }
+
+ @Override
+ public String getBucketId(Integer element, Context context) {
+ return Integer.toString(element % numBuckets);
+ }
+
+ @Override
+ public SimpleVersionedSerializer<String> getSerializer() {
+ return SimpleVersionedStringSerializer.INSTANCE;
+ }
+ }
+
+ /**
+ * Verifies the files written by the sink contains the expected integer sequences.
+ * The integers are partition into different buckets according to module, and each
+ * integer will be repeated by <tt>numSources</tt> times.
+ *
+ * @param path The directory to check.
+ * @param numRecords The total number of records.
+ * @param numBuckets The number of buckets to assign.
+ * @param numSources The parallelism of sources generating the sequences. Each integer will be
+ * repeat for <tt>numSources</tt> times.
+ */
+ public static void checkIntegerSequenceSinkOutput(String path, int numRecords, int numBuckets, int numSources) throws Exception {
+ File dir = new File(path);
+ String[] subDirNames = dir.list();
+ assertNotNull(subDirNames);
+
+ Arrays.sort(subDirNames, Comparator.comparingInt(Integer::parseInt));
+ assertEquals(numBuckets, subDirNames.length);
+ for (int i = 0; i < numBuckets; ++i) {
+ assertEquals(Integer.toString(i), subDirNames[i]);
+
+ // now check its content
+ File bucketDir = new File(path, subDirNames[i]);
+ assertTrue(
+ bucketDir.getAbsolutePath() + " Should be a existing directory",
+ bucketDir.isDirectory());
+
+ Map<Integer, Integer> counts = new HashMap<>();
+ File[] files = bucketDir.listFiles(f -> !f.getName().startsWith("."));
+ assertNotNull(files);
+
+ for (File file : files) {
+ assertTrue(file.isFile());
+
+ try (DataInputStream dataInputStream = new DataInputStream(new FileInputStream(file))) {
+ while (true) {
+ int value = dataInputStream.readInt();
+ counts.compute(value, (k, v) -> v == null ? 1 : v + 1);
+ }
+ } catch (EOFException e) {
+ // End the reading
+ }
+ }
+
+ int expectedCount = numRecords / numBuckets +
+ (i < numRecords % numBuckets ? 1 : 0);
+ assertEquals(expectedCount, counts.size());
+
+ for (int j = i; j < numRecords; j += numBuckets) {
+ assertEquals(
+ "The record " + j + " should occur " + numSources + " times, " +
+ " but only occurs " + counts.getOrDefault(j, 0) + "time",
+ numSources,
+ counts.getOrDefault(j, 0).intValue());
+ }
+ }
+ }
+}