This is an automated email from the ASF dual-hosted git repository. sewen pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 59055b55a0d384b92b7beb09f96c4ba08e9920b7 Author: Stephan Ewen <se...@apache.org> AuthorDate: Sun Oct 25 21:47:16 2020 +0100 [FLINK-19803][connector files] Make PendingSplitCheckpoint and its Serializer generic to support sub-classes of FileSourceSplit --- .../flink/connector/file/src/FileSource.java | 14 ++++---- .../file/src/PendingSplitsCheckpoint.java | 31 ++++++++++-------- .../src/PendingSplitsCheckpointSerializer.java | 38 +++++++++++++--------- .../src/impl/ContinuousFileSplitEnumerator.java | 4 +-- .../file/src/impl/StaticFileSplitEnumerator.java | 4 +-- .../src/PendingSplitsCheckpointSerializerTest.java | 33 +++++++++++-------- 6 files changed, 69 insertions(+), 55 deletions(-) diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/FileSource.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/FileSource.java index 6580742..ccfe8b0 100644 --- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/FileSource.java +++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/FileSource.java @@ -107,7 +107,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull; * @param <T> The type of the events/records produced by this source. */ @PublicEvolving -public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSplitsCheckpoint>, ResultTypeQueryable<T> { +public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSplitsCheckpoint<FileSourceSplit>>, ResultTypeQueryable<T> { private static final long serialVersionUID = 1L; @@ -177,7 +177,7 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp } @Override - public SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint> createEnumerator( + public SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint<FileSourceSplit>> createEnumerator( SplitEnumeratorContext<FileSourceSplit> enumContext) { final FileEnumerator enumerator = enumeratorFactory.create(); @@ -195,9 +195,9 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp } @Override - public SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint> restoreEnumerator( + public SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint<FileSourceSplit>> restoreEnumerator( SplitEnumeratorContext<FileSourceSplit> enumContext, - PendingSplitsCheckpoint checkpoint) throws IOException { + PendingSplitsCheckpoint<FileSourceSplit> checkpoint) throws IOException { final FileEnumerator enumerator = enumeratorFactory.create(); @@ -210,8 +210,8 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp } @Override - public SimpleVersionedSerializer<PendingSplitsCheckpoint> getEnumeratorCheckpointSerializer() { - return PendingSplitsCheckpointSerializer.INSTANCE; + public SimpleVersionedSerializer<PendingSplitsCheckpoint<FileSourceSplit>> getEnumeratorCheckpointSerializer() { + return new PendingSplitsCheckpointSerializer<>(getSplitSerializer()); } @Override @@ -223,7 +223,7 @@ public final class FileSource<T> implements Source<T, FileSourceSplit, PendingSp // helpers // ------------------------------------------------------------------------ - private SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint> createSplitEnumerator( + private SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint<FileSourceSplit>> createSplitEnumerator( SplitEnumeratorContext<FileSourceSplit> context, FileEnumerator enumerator, Collection<FileSourceSplit> splits, diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/PendingSplitsCheckpoint.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/PendingSplitsCheckpoint.java index 1515690..847f99e 100644 --- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/PendingSplitsCheckpoint.java +++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/PendingSplitsCheckpoint.java @@ -33,10 +33,10 @@ import static org.apache.flink.util.Preconditions.checkNotNull; * A checkpoint of the current state of the containing the currently pending splits that are not yet assigned. */ @PublicEvolving -public final class PendingSplitsCheckpoint { +public final class PendingSplitsCheckpoint<SplitT extends FileSourceSplit> { /** The splits in the checkpoint. */ - private final Collection<FileSourceSplit> splits; + private final Collection<SplitT> splits; /** The paths that are no longer in the enumerator checkpoint, but have been processed * before and should this be ignored. Relevant only for sources in continuous monitoring mode. */ @@ -48,14 +48,14 @@ public final class PendingSplitsCheckpoint { @Nullable byte[] serializedFormCache; - private PendingSplitsCheckpoint(Collection<FileSourceSplit> splits, Collection<Path> alreadyProcessedPaths) { + private PendingSplitsCheckpoint(Collection<SplitT> splits, Collection<Path> alreadyProcessedPaths) { this.splits = Collections.unmodifiableCollection(splits); this.alreadyProcessedPaths = Collections.unmodifiableCollection(alreadyProcessedPaths); } // ------------------------------------------------------------------------ - public Collection<FileSourceSplit> getSplits() { + public Collection<SplitT> getSplits() { return splits; } @@ -67,27 +67,30 @@ public final class PendingSplitsCheckpoint { // factories // ------------------------------------------------------------------------ - public static PendingSplitsCheckpoint fromCollectionSnapshot(Collection<FileSourceSplit> splits) { + public static <T extends FileSourceSplit> PendingSplitsCheckpoint<T> fromCollectionSnapshot( + final Collection<T> splits) { checkNotNull(splits); // create a copy of the collection to make sure this checkpoint is immutable - final Collection<FileSourceSplit> copy = new ArrayList<>(splits); - return new PendingSplitsCheckpoint(copy, Collections.emptySet()); + final Collection<T> copy = new ArrayList<>(splits); + return new PendingSplitsCheckpoint<>(copy, Collections.emptySet()); } - public static PendingSplitsCheckpoint fromCollectionSnapshot( - Collection<FileSourceSplit> splits, - Collection<Path> alreadyProcessedPaths) { + public static <T extends FileSourceSplit> PendingSplitsCheckpoint<T> fromCollectionSnapshot( + final Collection<T> splits, + final Collection<Path> alreadyProcessedPaths) { checkNotNull(splits); // create a copy of the collection to make sure this checkpoint is immutable - final Collection<FileSourceSplit> splitsCopy = new ArrayList<>(splits); + final Collection<T> splitsCopy = new ArrayList<>(splits); final Collection<Path> pathsCopy = new ArrayList<>(alreadyProcessedPaths); - return new PendingSplitsCheckpoint(splitsCopy, pathsCopy); + return new PendingSplitsCheckpoint<>(splitsCopy, pathsCopy); } - static PendingSplitsCheckpoint reusingCollection(Collection<FileSourceSplit> splits, Collection<Path> alreadyProcessedPaths) { - return new PendingSplitsCheckpoint(splits, alreadyProcessedPaths); + static <T extends FileSourceSplit> PendingSplitsCheckpoint<T> reusingCollection( + final Collection<T> splits, + final Collection<Path> alreadyProcessedPaths) { + return new PendingSplitsCheckpoint<>(splits, alreadyProcessedPaths); } } diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/PendingSplitsCheckpointSerializer.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/PendingSplitsCheckpointSerializer.java index e87ca7d..9721a53 100644 --- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/PendingSplitsCheckpointSerializer.java +++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/PendingSplitsCheckpointSerializer.java @@ -30,19 +30,25 @@ import java.util.ArrayList; import java.util.Collection; import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkNotNull; /** * A serializer for the {@link PendingSplitsCheckpoint}. */ @PublicEvolving -public final class PendingSplitsCheckpointSerializer implements SimpleVersionedSerializer<PendingSplitsCheckpoint> { - - public static final PendingSplitsCheckpointSerializer INSTANCE = new PendingSplitsCheckpointSerializer(); +public final class PendingSplitsCheckpointSerializer<T extends FileSourceSplit> + implements SimpleVersionedSerializer<PendingSplitsCheckpoint<T>> { private static final int VERSION = 1; private static final int VERSION_1_MAGIC_NUMBER = 0xDEADBEEF; + private final SimpleVersionedSerializer<T> splitSerializer; + + public PendingSplitsCheckpointSerializer(SimpleVersionedSerializer<T> splitSerializer) { + this.splitSerializer = checkNotNull(splitSerializer); + } + // ------------------------------------------------------------------------ @Override @@ -51,7 +57,7 @@ public final class PendingSplitsCheckpointSerializer implements SimpleVersionedS } @Override - public byte[] serialize(PendingSplitsCheckpoint checkpoint) throws IOException { + public byte[] serialize(PendingSplitsCheckpoint<T> checkpoint) throws IOException { checkArgument(checkpoint.getClass() == PendingSplitsCheckpoint.class, "Cannot serialize subclasses of PendingSplitsCheckpoint"); @@ -60,17 +66,17 @@ public final class PendingSplitsCheckpointSerializer implements SimpleVersionedS return checkpoint.serializedFormCache; } - final FileSourceSplitSerializer serializer = FileSourceSplitSerializer.INSTANCE; - final Collection<FileSourceSplit> splits = checkpoint.getSplits(); + final SimpleVersionedSerializer<T> splitSerializer = this.splitSerializer; // stack cache + final Collection<T> splits = checkpoint.getSplits(); final Collection<Path> processedPaths = checkpoint.getAlreadyProcessedPaths(); final ArrayList<byte[]> serializedSplits = new ArrayList<>(splits.size()); final ArrayList<byte[]> serializedPaths = new ArrayList<>(processedPaths.size()); - int totalLen = 16; // four ints: magic, version, count splits, count paths + int totalLen = 16; // four ints: magic, version of split serializer, count splits, count paths - for (FileSourceSplit split : splits) { - final byte[] serSplit = serializer.serialize(split); + for (T split : splits) { + final byte[] serSplit = splitSerializer.serialize(split); serializedSplits.add(serSplit); totalLen += serSplit.length + 4; // 4 bytes for the length field } @@ -84,7 +90,7 @@ public final class PendingSplitsCheckpointSerializer implements SimpleVersionedS final byte[] result = new byte[totalLen]; final ByteBuffer byteBuffer = ByteBuffer.wrap(result).order(ByteOrder.LITTLE_ENDIAN); byteBuffer.putInt(VERSION_1_MAGIC_NUMBER); - byteBuffer.putInt(serializer.getVersion()); + byteBuffer.putInt(splitSerializer.getVersion()); byteBuffer.putInt(serializedSplits.size()); byteBuffer.putInt(serializedPaths.size()); @@ -107,14 +113,14 @@ public final class PendingSplitsCheckpointSerializer implements SimpleVersionedS } @Override - public PendingSplitsCheckpoint deserialize(int version, byte[] serialized) throws IOException { + public PendingSplitsCheckpoint<T> deserialize(int version, byte[] serialized) throws IOException { if (version == 1) { return deserializeV1(serialized); } throw new IOException("Unknown version: " + version); } - private static PendingSplitsCheckpoint deserializeV1(byte[] serialized) throws IOException { + private PendingSplitsCheckpoint<T> deserializeV1(byte[] serialized) throws IOException { final ByteBuffer bb = ByteBuffer.wrap(serialized).order(ByteOrder.LITTLE_ENDIAN); final int magic = bb.getInt(); @@ -123,18 +129,18 @@ public final class PendingSplitsCheckpointSerializer implements SimpleVersionedS "Expected: %X , found %X", VERSION_1_MAGIC_NUMBER, magic)); } - final int version = bb.getInt(); + final int splitSerializerVersion = bb.getInt(); final int numSplits = bb.getInt(); final int numPaths = bb.getInt(); - final FileSourceSplitSerializer serializer = FileSourceSplitSerializer.INSTANCE; - final ArrayList<FileSourceSplit> splits = new ArrayList<>(numSplits); + final SimpleVersionedSerializer<T> splitSerializer = this.splitSerializer; // stack cache + final ArrayList<T> splits = new ArrayList<>(numSplits); final ArrayList<Path> paths = new ArrayList<>(numPaths); for (int remaining = numSplits; remaining > 0; remaining--) { final byte[] bytes = new byte[bb.getInt()]; bb.get(bytes); - final FileSourceSplit split = serializer.deserialize(version, bytes); + final T split = splitSerializer.deserialize(splitSerializerVersion, bytes); splits.add(split); } diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/ContinuousFileSplitEnumerator.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/ContinuousFileSplitEnumerator.java index c6a6346..59ccece 100644 --- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/ContinuousFileSplitEnumerator.java +++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/ContinuousFileSplitEnumerator.java @@ -49,7 +49,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull; * A continuously monitoring enumerator. */ @Internal -public class ContinuousFileSplitEnumerator implements SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint> { +public class ContinuousFileSplitEnumerator implements SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint<FileSourceSplit>> { private static final Logger LOG = LoggerFactory.getLogger(ContinuousFileSplitEnumerator.class); @@ -123,7 +123,7 @@ public class ContinuousFileSplitEnumerator implements SplitEnumerator<FileSource } @Override - public PendingSplitsCheckpoint snapshotState() throws Exception { + public PendingSplitsCheckpoint<FileSourceSplit> snapshotState() throws Exception { return PendingSplitsCheckpoint.fromCollectionSnapshot(splitAssigner.remainingSplits(), pathsAlreadyProcessed); } diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StaticFileSplitEnumerator.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StaticFileSplitEnumerator.java index 15c20f1..f194dc4 100644 --- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StaticFileSplitEnumerator.java +++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/StaticFileSplitEnumerator.java @@ -52,7 +52,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull; * {@link FileEnumerator} and in {@link FileSplitAssigner}, respectively. */ @Internal -public class StaticFileSplitEnumerator implements SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint> { +public class StaticFileSplitEnumerator implements SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint<FileSourceSplit>> { private static final Logger LOG = LoggerFactory.getLogger(StaticFileSplitEnumerator.class); @@ -102,7 +102,7 @@ public class StaticFileSplitEnumerator implements SplitEnumerator<FileSourceSpli } @Override - public PendingSplitsCheckpoint snapshotState() { + public PendingSplitsCheckpoint<FileSourceSplit> snapshotState() { return PendingSplitsCheckpoint.fromCollectionSnapshot(splitAssigner.remainingSplits()); } diff --git a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/PendingSplitsCheckpointSerializerTest.java b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/PendingSplitsCheckpointSerializerTest.java index bc89dce..092c21c 100644 --- a/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/PendingSplitsCheckpointSerializerTest.java +++ b/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/PendingSplitsCheckpointSerializerTest.java @@ -41,54 +41,54 @@ public class PendingSplitsCheckpointSerializerTest { @Test public void serializeEmptyCheckpoint() throws Exception { - final PendingSplitsCheckpoint checkpoint = + final PendingSplitsCheckpoint<FileSourceSplit> checkpoint = PendingSplitsCheckpoint.fromCollectionSnapshot(Collections.emptyList()); - final PendingSplitsCheckpoint deSerialized = serializeAndDeserialize(checkpoint); + final PendingSplitsCheckpoint<FileSourceSplit> deSerialized = serializeAndDeserialize(checkpoint); assertCheckpointsEqual(checkpoint, deSerialized); } @Test public void serializeSomeSplits() throws Exception { - final PendingSplitsCheckpoint checkpoint = PendingSplitsCheckpoint.fromCollectionSnapshot( + final PendingSplitsCheckpoint<FileSourceSplit> checkpoint = PendingSplitsCheckpoint.fromCollectionSnapshot( Arrays.asList(testSplit1(), testSplit2(), testSplit3())); - final PendingSplitsCheckpoint deSerialized = serializeAndDeserialize(checkpoint); + final PendingSplitsCheckpoint<FileSourceSplit> deSerialized = serializeAndDeserialize(checkpoint); assertCheckpointsEqual(checkpoint, deSerialized); } @Test public void serializeSplitsAndProcessedPaths() throws Exception { - final PendingSplitsCheckpoint checkpoint = PendingSplitsCheckpoint.fromCollectionSnapshot( + final PendingSplitsCheckpoint<FileSourceSplit> checkpoint = PendingSplitsCheckpoint.fromCollectionSnapshot( Arrays.asList(testSplit1(), testSplit2(), testSplit3()), Arrays.asList(new Path("file:/some/path"), new Path("s3://bucket/key/and/path"), new Path("hdfs://namenode:12345/path"))); - final PendingSplitsCheckpoint deSerialized = serializeAndDeserialize(checkpoint); + final PendingSplitsCheckpoint<FileSourceSplit> deSerialized = serializeAndDeserialize(checkpoint); assertCheckpointsEqual(checkpoint, deSerialized); } @Test public void repeatedSerialization() throws Exception { - final PendingSplitsCheckpoint checkpoint = PendingSplitsCheckpoint.fromCollectionSnapshot( + final PendingSplitsCheckpoint<FileSourceSplit> checkpoint = PendingSplitsCheckpoint.fromCollectionSnapshot( Arrays.asList(testSplit3(), testSplit1())); serializeAndDeserialize(checkpoint); serializeAndDeserialize(checkpoint); - final PendingSplitsCheckpoint deSerialized = serializeAndDeserialize(checkpoint); + final PendingSplitsCheckpoint<FileSourceSplit> deSerialized = serializeAndDeserialize(checkpoint); assertCheckpointsEqual(checkpoint, deSerialized); } @Test public void repeatedSerializationCaches() throws Exception { - final PendingSplitsCheckpoint checkpoint = PendingSplitsCheckpoint.fromCollectionSnapshot( + final PendingSplitsCheckpoint<FileSourceSplit> checkpoint = PendingSplitsCheckpoint.fromCollectionSnapshot( Collections.singletonList(testSplit2())); - final byte[] ser1 = PendingSplitsCheckpointSerializer.INSTANCE.serialize(checkpoint); - final byte[] ser2 = PendingSplitsCheckpointSerializer.INSTANCE.serialize(checkpoint); + final byte[] ser1 = new PendingSplitsCheckpointSerializer<>(FileSourceSplitSerializer.INSTANCE).serialize(checkpoint); + final byte[] ser2 = new PendingSplitsCheckpointSerializer<>(FileSourceSplitSerializer.INSTANCE).serialize(checkpoint); assertSame(ser1, ser2); } @@ -122,13 +122,18 @@ public class PendingSplitsCheckpointSerializerTest { 1234567); } - private static PendingSplitsCheckpoint serializeAndDeserialize(PendingSplitsCheckpoint split) throws IOException { - final PendingSplitsCheckpointSerializer serializer = new PendingSplitsCheckpointSerializer(); + private static PendingSplitsCheckpoint<FileSourceSplit> serializeAndDeserialize( + final PendingSplitsCheckpoint<FileSourceSplit> split) throws IOException { + + final PendingSplitsCheckpointSerializer<FileSourceSplit> serializer = + new PendingSplitsCheckpointSerializer<>(FileSourceSplitSerializer.INSTANCE); final byte[] bytes = SimpleVersionedSerialization.writeVersionAndSerialize(serializer, split); return SimpleVersionedSerialization.readVersionAndDeSerialize(serializer, bytes); } - private static void assertCheckpointsEqual(PendingSplitsCheckpoint expected, PendingSplitsCheckpoint actual) { + private static void assertCheckpointsEqual( + final PendingSplitsCheckpoint<FileSourceSplit> expected, + final PendingSplitsCheckpoint<FileSourceSplit> actual) { assertOrderedCollectionEquals(expected.getSplits(), actual.getSplits(), FileSourceSplitSerializerTest::assertSplitsEqual); assertOrderedCollectionEquals(