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/05/18 14:12:36 UTC
[flink] 02/04: [FLINK-17593][Connectors/FileSystem] Turn
BucketStateSerializerTest into an upgrade test
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f1e9362cb09e094f1ea6d1984dc65ea34443bb5b
Author: GuoWei Ma <gu...@gmail.com>
AuthorDate: Mon May 18 15:19:07 2020 +0800
[FLINK-17593][Connectors/FileSystem] Turn BucketStateSerializerTest into an upgrade test
---
.../sink/filesystem/BucketStateSerializerTest.java | 424 +++++++++++++--------
.../bucket-state-migration-test/empty-v1/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.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.a70190d6-d080-43a8-b414-746b09d3a8a0 | 1 +
.../only-in-progress-v1/snapshot | Bin 0 -> 404 bytes
17 files changed, 282 insertions(+), 164 deletions(-)
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 f2c1f8b..81c5766 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
@@ -18,253 +18,349 @@
package org.apache.flink.streaming.api.functions.sink.filesystem;
-import org.apache.flink.core.fs.FileStatus;
+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.fs.RecoverableFsDataOutputStream;
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.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 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.File;
import java.io.IOException;
-import java.nio.charset.Charset;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
+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 BucketStateSerializer}.
+ * Tests for the {@link BucketStateSerializer} that verify we can still read snapshots written using
+ * an older version of the serializer. 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.
*/
+@RunWith(Parameterized.class)
public class BucketStateSerializerTest {
+ private static final int CURRENT_VERSION = 1;
+
+ @Parameterized.Parameters(name = "Previous Version = {0}")
+ public static Collection<Integer> previousVersions() {
+ return Arrays.asList(1);
+ }
+
+ @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";
+
@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");
+ }
+
@Test
- public void testSerializationEmpty() throws IOException {
- final File testFolder = tempFolder.newFolder();
- final FileSystem fs = FileSystem.get(testFolder.toURI());
- final RecoverableWriter writer = fs.createRecoverableWriter();
+ @Ignore
+ public void prepareDeserializationEmpty() throws IOException {
- final Path testBucket = new Path(testFolder.getPath(), "test");
+ final String scenarioName = "empty";
+ final java.nio.file.Path scenarioPath = getResourcePath(scenarioName, CURRENT_VERSION);
- final BucketState<String> bucketState = new BucketState<>(
- "test", testBucket, Long.MAX_VALUE, null, new HashMap<>());
+ FileUtils.deleteDirectory(scenarioPath.toFile());
+ Files.createDirectories(scenarioPath);
- final SimpleVersionedSerializer<BucketState<String>> serializer =
- new BucketStateSerializer<>(
- writer.getResumeRecoverableSerializer(),
- writer.getCommitRecoverableSerializer(),
- SimpleVersionedStringSerializer.INSTANCE
- );
+ final java.nio.file.Path outputPath = getOutputPath(scenarioName, CURRENT_VERSION);
+ final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
- byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(serializer, bucketState);
- final BucketState<String> recoveredState = SimpleVersionedSerialization.readVersionAndDeSerialize(serializer, bytes);
+ final Bucket<String, String> bucket =
+ createNewBucket(testBucketPath);
- Assert.assertEquals(testBucket, recoveredState.getBucketPath());
- Assert.assertNull(recoveredState.getInProgressResumableFile());
- Assert.assertTrue(recoveredState.getCommittableFilesPerCheckpoint().isEmpty());
+ final BucketState<String> bucketState = bucket.onReceptionOfCheckpoint(0);
+
+ byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(
+ bucketStateSerializer(),
+ bucketState);
+ Files.write(getSnapshotPath(scenarioName, CURRENT_VERSION), bytes);
}
@Test
- public void testSerializationOnlyInProgress() throws IOException {
- final File testFolder = tempFolder.newFolder();
- final FileSystem fs = FileSystem.get(testFolder.toURI());
+ public void testSerializationEmpty() throws IOException {
- final Path testBucket = new Path(testFolder.getPath(), "test");
+ final String scenarioName = "empty";
+ final java.nio.file.Path outputPath = getOutputPath(scenarioName, previousVersion);
+ final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
+ final BucketState<String> recoveredState = readBucketState(scenarioName, previousVersion);
- final RecoverableWriter writer = fs.createRecoverableWriter();
- final RecoverableFsDataOutputStream stream = writer.open(testBucket);
- stream.write(IN_PROGRESS_CONTENT.getBytes(Charset.forName("UTF-8")));
+ final Bucket<String, String> bucket = restoreBucket(0, recoveredState);
- final RecoverableWriter.ResumeRecoverable current = stream.persist();
+ Assert.assertEquals(testBucketPath, bucket.getBucketPath());
+ Assert.assertNull(bucket.getInProgressPart());
+ Assert.assertTrue(bucket.getPendingPartsPerCheckpoint().isEmpty());
+ }
- final BucketState<String> bucketState = new BucketState<>(
- "test", testBucket, Long.MAX_VALUE, current, new HashMap<>());
+ @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 SimpleVersionedSerializer<BucketState<String>> serializer =
- new BucketStateSerializer<>(
- writer.getResumeRecoverableSerializer(),
- writer.getCommitRecoverableSerializer(),
- SimpleVersionedStringSerializer.INSTANCE
- );
+ final java.nio.file.Path outputPath = getOutputPath(scenarioName, CURRENT_VERSION);
+ final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
- final byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(serializer, bucketState);
+ final Bucket<String, String> bucket =
+ createNewBucket(testBucketPath);
- // to simulate that everything is over for file.
- stream.close();
+ bucket.write(IN_PROGRESS_CONTENT, System.currentTimeMillis());
- final BucketState<String> recoveredState = SimpleVersionedSerialization.readVersionAndDeSerialize(serializer, bytes);
+ final BucketState<String> bucketState = bucket.onReceptionOfCheckpoint(0);
- Assert.assertEquals(testBucket, recoveredState.getBucketPath());
+ final byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(
+ bucketStateSerializer(), bucketState);
- FileStatus[] statuses = fs.listStatus(testBucket.getParent());
- Assert.assertEquals(1L, statuses.length);
- Assert.assertTrue(
- statuses[0].getPath().getPath().startsWith(
- (new Path(testBucket.getParent(), ".test.inprogress")).getPath())
- );
+ Files.write(getSnapshotPath(scenarioName, CURRENT_VERSION), bytes);
}
@Test
- public void testSerializationFull() throws IOException {
- final int noOfTasks = 5;
+ public void testSerializationOnlyInProgress() throws IOException {
- final File testFolder = tempFolder.newFolder();
- final FileSystem fs = FileSystem.get(testFolder.toURI());
- final RecoverableWriter writer = fs.createRecoverableWriter();
+ final String scenarioName = "only-in-progress";
+ final java.nio.file.Path outputPath = getOutputPath(scenarioName, previousVersion);
- final Path bucketPath = new Path(testFolder.getPath());
+ final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
- // pending for checkpoints
- final Map<Long, List<RecoverableWriter.CommitRecoverable>> commitRecoverables = new HashMap<>();
- for (int i = 0; i < noOfTasks; i++) {
- final List<RecoverableWriter.CommitRecoverable> recoverables = new ArrayList<>();
- for (int j = 0; j < 2 + i; j++) {
- final Path part = new Path(bucketPath, "part-" + i + '-' + j);
-
- final RecoverableFsDataOutputStream stream = writer.open(part);
- stream.write((PENDING_CONTENT + '-' + j).getBytes(Charset.forName("UTF-8")));
- recoverables.add(stream.closeForCommit().getRecoverable());
- }
- commitRecoverables.put((long) i, recoverables);
- }
+ final BucketState<String> recoveredState = readBucketState(scenarioName, previousVersion);
- // in-progress
- final Path testBucket = new Path(bucketPath, "test-2");
- final RecoverableFsDataOutputStream stream = writer.open(testBucket);
- stream.write(IN_PROGRESS_CONTENT.getBytes(Charset.forName("UTF-8")));
+ final Bucket<String, String> bucket = restoreBucket(0, recoveredState);
- final RecoverableWriter.ResumeRecoverable current = stream.persist();
+ Assert.assertEquals(testBucketPath, bucket.getBucketPath());
- final BucketState<String> bucketState = new BucketState<>(
- "test-2", bucketPath, Long.MAX_VALUE, current, commitRecoverables);
- final SimpleVersionedSerializer<BucketState<String>> serializer =
- new BucketStateSerializer<>(
- writer.getResumeRecoverableSerializer(),
- writer.getCommitRecoverableSerializer(),
- SimpleVersionedStringSerializer.INSTANCE
- );
- stream.close();
+ //check restore the correct in progress file writer
+ Assert.assertEquals(8, bucket.getInProgressPart().getSize());
- byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(serializer, bucketState);
+ long numFiles = Files.list(Paths.get(testBucketPath.toString()))
+ .map(file -> {
+ assertThat(
+ file.getFileName().toString(),
+ startsWith(".part-0-0.inprogress"));
+ return 1;
+ })
+ .count();
- final BucketState<String> recoveredState = SimpleVersionedSerialization.readVersionAndDeSerialize(serializer, bytes);
+ assertThat(numFiles, is(1L));
+ }
- Assert.assertEquals(bucketPath, recoveredState.getBucketPath());
+ @Test
+ @Ignore
+ public void prepareDeserializationFull() throws IOException {
+ prepareDeserializationFull(true, "full");
+ }
- final Map<Long, List<RecoverableWriter.CommitRecoverable>> recoveredRecoverables = recoveredState.getCommittableFilesPerCheckpoint();
- Assert.assertEquals(5L, recoveredRecoverables.size());
+ @Test
+ public void testSerializationFull() throws IOException {
+ testDeserializationFull(true, "full");
+ }
- // recover and commit
- for (Map.Entry<Long, List<RecoverableWriter.CommitRecoverable>> entry: recoveredRecoverables.entrySet()) {
- for (RecoverableWriter.CommitRecoverable recoverable: entry.getValue()) {
- writer.recoverForCommit(recoverable).commit();
- }
- }
+ @Test
+ @Ignore
+ public void prepareDeserializationNullInProgress() throws IOException {
+ prepareDeserializationFull(false, "full-no-in-progress");
+ }
- FileStatus[] filestatuses = fs.listStatus(bucketPath);
- Set<String> paths = new HashSet<>(filestatuses.length);
- for (FileStatus filestatus : filestatuses) {
- paths.add(filestatus.getPath().getPath());
- }
+ @Test
+ public void testSerializationNullInProgress() throws IOException {
+ testDeserializationFull(false, "full-no-in-progress");
+ }
- for (int i = 0; i < noOfTasks; i++) {
- for (int j = 0; j < 2 + i; j++) {
- final String part = new Path(bucketPath, "part-" + i + '-' + j).getPath();
- Assert.assertTrue(paths.contains(part));
- paths.remove(part);
- }
- }
+ private static void prepareDeserializationFull(final boolean withInProgress, final String scenarioName) throws IOException {
- // only the in-progress must be left
- Assert.assertEquals(1L, paths.size());
+ final java.nio.file.Path scenarioPath = getResourcePath(scenarioName, CURRENT_VERSION);
+ FileUtils.deleteDirectory(Paths.get(scenarioPath.toString() + "-template").toFile());
+ Files.createDirectories(scenarioPath);
- // verify that the in-progress file is still there
- Assert.assertTrue(paths.iterator().next().startsWith(
- (new Path(testBucket.getParent(), ".test-2.inprogress").getPath())));
- }
+ final int noOfPendingCheckpoints = 5;
- @Test
- public void testSerializationNullInProgress() throws IOException {
- final int noOfTasks = 5;
+ final java.nio.file.Path outputPath = getOutputPath(scenarioName, CURRENT_VERSION);
- final File testFolder = tempFolder.newFolder();
- final FileSystem fs = FileSystem.get(testFolder.toURI());
- final RecoverableWriter writer = fs.createRecoverableWriter();
+ final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
- final Path bucketPath = new Path(testFolder.getPath());
+ final Bucket<String, String> bucket = createNewBucket(testBucketPath);
+ BucketState<String> bucketState = null;
// pending for checkpoints
- final Map<Long, List<RecoverableWriter.CommitRecoverable>> commitRecoverables = new HashMap<>();
- for (int i = 0; i < noOfTasks; i++) {
- final List<RecoverableWriter.CommitRecoverable> recoverables = new ArrayList<>();
- for (int j = 0; j < 2 + i; j++) {
- final Path part = new Path(bucketPath, "test-" + i + '-' + j);
-
- final RecoverableFsDataOutputStream stream = writer.open(part);
- stream.write((PENDING_CONTENT + '-' + j).getBytes(Charset.forName("UTF-8")));
- recoverables.add(stream.closeForCommit().getRecoverable());
- }
- commitRecoverables.put((long) i, recoverables);
+ 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 RecoverableWriter.ResumeRecoverable current = null;
+ final byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(bucketStateSerializer(), bucketState);
- final BucketState<String> bucketState = new BucketState<>(
- "", bucketPath, Long.MAX_VALUE, current, commitRecoverables);
+ Files.write(getSnapshotPath(scenarioName, CURRENT_VERSION), bytes);
- final SimpleVersionedSerializer<BucketState<String>> serializer = new BucketStateSerializer<>(
- writer.getResumeRecoverableSerializer(),
- writer.getCommitRecoverableSerializer(),
- SimpleVersionedStringSerializer.INSTANCE
- );
+ // 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);
+ }
- byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(serializer, bucketState);
+ private void testDeserializationFull(final boolean withInProgress, final String scenarioName) throws IOException {
- final BucketState<String> recoveredState = SimpleVersionedSerialization.readVersionAndDeSerialize(serializer, bytes);
+ try {
+ final java.nio.file.Path outputPath = getOutputPath(scenarioName, previousVersion);
+ final Path testBucketPath = new Path(outputPath.resolve(BUCKET_ID).toString());
+ // restore the state
+ final BucketState<String> recoveredState = readBucketStateFromTemplate(scenarioName, previousVersion);
+ final int noOfPendingCheckpoints = 5;
- Assert.assertEquals(bucketPath, recoveredState.getBucketPath());
- Assert.assertNull(recoveredState.getInProgressResumableFile());
+ // there are 5 checkpoint does not complete.
+ final Map<Long, List<RecoverableWriter.CommitRecoverable>>
+ pendingFileRecoverables = recoveredState.getCommittableFilesPerCheckpoint();
+ Assert.assertEquals(5L, pendingFileRecoverables.size());
- final Map<Long, List<RecoverableWriter.CommitRecoverable>> recoveredRecoverables = recoveredState.getCommittableFilesPerCheckpoint();
- Assert.assertEquals(5L, recoveredRecoverables.size());
+ final Set<String> beforeRestorePaths = Files.list(outputPath.resolve(BUCKET_ID))
+ .map(file -> file.getFileName().toString())
+ .collect(Collectors.toSet());
- // recover and commit
- for (Map.Entry<Long, List<RecoverableWriter.CommitRecoverable>> entry: recoveredRecoverables.entrySet()) {
- for (RecoverableWriter.CommitRecoverable recoverable: entry.getValue()) {
- writer.recoverForCommit(recoverable).commit();
+ // 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)));
}
- }
- FileStatus[] filestatuses = fs.listStatus(bucketPath);
- Set<String> paths = new HashSet<>(filestatuses.length);
- for (FileStatus filestatus : filestatuses) {
- paths.add(filestatus.getPath().getPath());
- }
+ // recover and commit
+ final Bucket bucket = restoreBucket(noOfPendingCheckpoints + 1, recoveredState);
+ Assert.assertEquals(testBucketPath, bucket.getBucketPath());
+ Assert.assertEquals(0, bucket.getPendingPartsPerCheckpoint().size());
+
+ 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));
- for (int i = 0; i < noOfTasks; i++) {
- for (int j = 0; j < 2 + i; j++) {
- final String part = new Path(bucketPath, "test-" + i + '-' + j).getPath();
- Assert.assertTrue(paths.contains(part));
- paths.remove(part);
+ // verify that the in-progress file is still there
+ assertThat(afterRestorePaths, hasItem(startsWith(".part-0-" + noOfPendingCheckpoints + ".inprogress")));
+ } else {
+ assertThat(afterRestorePaths, empty());
}
+ } finally {
+ FileUtils.deleteDirectory(getResourcePath(scenarioName, previousVersion).toFile());
}
+ }
+
+ private static Bucket<String, String> createNewBucket(final Path bucketPath) throws IOException {
+ return Bucket.getNew(
+ FileSystem.getLocalFileSystem().createRecoverableWriter(),
+ 0,
+ BUCKET_ID,
+ bucketPath,
+ 0,
+ new RowWisePartWriter.Factory<>(new SimpleStringEncoder<>()),
+ DefaultRollingPolicy.builder().withMaxPartSize(10).build(),
+ OutputFileConfig.builder().build());
+ }
+
+ private static Bucket<String, String> restoreBucket(final int initialPartCounter, final BucketState<String> bucketState) throws IOException {
+ return Bucket.restore(
+ FileSystem.getLocalFileSystem().createRecoverableWriter(),
+ 0,
+ initialPartCounter,
+ new RowWisePartWriter.Factory<>(new SimpleStringEncoder<>()),
+ DefaultRollingPolicy.builder().withMaxPartSize(10).build(),
+ bucketState,
+ OutputFileConfig.builder().build());
+ }
+
+ private static BucketState<String> readBucketState(final String scenarioName, final int version) throws IOException {
+ byte[] bytes = Files.readAllBytes(getSnapshotPath(scenarioName, version));
+ 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);
+
+ // 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);
+ }
+
+ private static SimpleVersionedSerializer<BucketState<String>> bucketStateSerializer() throws IOException {
+ RecoverableWriter recoverableWriter = FileSystem.getLocalFileSystem().createRecoverableWriter();
+ return new BucketStateSerializer<>(
+ recoverableWriter.getResumeRecoverableSerializer(),
+ recoverableWriter.getCommitRecoverableSerializer(),
+ SimpleVersionedStringSerializer.INSTANCE);
+ }
- // only the in-progress must be left
- Assert.assertTrue(paths.isEmpty());
+ 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());
}
}
diff --git a/flink-streaming-java/src/test/resources/bucket-state-migration-test/empty-v1/snapshot b/flink-streaming-java/src/test/resources/bucket-state-migration-test/empty-v1/snapshot
new file mode 100644
index 0000000..9700f3d
Binary files /dev/null and b/flink-streaming-java/src/test/resources/bucket-state-migration-test/empty-v1/snapshot differ
diff --git a/flink-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/snapshot b/flink-streaming-java/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-streaming-java/src/test/resources/bucket-state-migration-test/full-no-in-progress-v1-template/snapshot differ
diff --git a/flink-streaming-java/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-0.inprogress.8fec17e9-5d54-4fa9-aebb-70736fe03c82 b/flink-streaming-java/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-streaming-java/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-streaming-java/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-1.inprogress.0035b171-2759-403a-8d6c-4612b28a7a6c b/flink-streaming-java/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-streaming-java/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-streaming-java/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-2.inprogress.49da8048-af6b-4665-b4f6-b659cb38dc97 b/flink-streaming-java/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-streaming-java/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-streaming-java/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-3.inprogress.d13ec4e0-07b5-4f4e-9be8-9fb457cbcde9 b/flink-streaming-java/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-streaming-java/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-streaming-java/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-4.inprogress.123ac2c7-f92a-476a-a848-1369b93d82a7 b/flink-streaming-java/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-streaming-java/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-streaming-java/src/test/resources/bucket-state-migration-test/full-v1-template/bucket/test-bucket/.part-0-5.inprogress.32f5a28f-20e1-48da-9951-10e795133d64 b/flink-streaming-java/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-streaming-java/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-streaming-java/src/test/resources/bucket-state-migration-test/full-v1-template/snapshot b/flink-streaming-java/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-streaming-java/src/test/resources/bucket-state-migration-test/full-v1-template/snapshot differ
diff --git a/flink-streaming-java/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-streaming-java/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-streaming-java/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-streaming-java/src/test/resources/bucket-state-migration-test/only-in-progress-v1/snapshot b/flink-streaming-java/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-streaming-java/src/test/resources/bucket-state-migration-test/only-in-progress-v1/snapshot differ