You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2020/05/16 15:25:14 UTC
[flink] 10/14: [FLINK-10740][DataStream API] Simplify
SourceOperator by using SimpleVersionedListState
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 61d0344cb973fd42c210d9a97f3ea1c0697445ce
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Sun May 10 20:50:29 2020 +0200
[FLINK-10740][DataStream API] Simplify SourceOperator by using SimpleVersionedListState
---
.../java/org/apache/flink/util/CollectionUtil.java | 17 +++++
.../streaming/api/operators/SourceOperator.java | 75 ++++------------------
.../api/operators/SourceOperatorTest.java | 20 ++----
3 files changed, 36 insertions(+), 76 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/CollectionUtil.java b/flink-core/src/main/java/org/apache/flink/util/CollectionUtil.java
index 45ceaa4..89d61f9 100644
--- a/flink-core/src/main/java/org/apache/flink/util/CollectionUtil.java
+++ b/flink-core/src/main/java/org/apache/flink/util/CollectionUtil.java
@@ -20,8 +20,11 @@ package org.apache.flink.util;
import org.apache.flink.annotation.Internal;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -82,4 +85,18 @@ public final class CollectionUtil {
.map(projector)
.collect(toList());
}
+
+ /**
+ * Collects the elements in the Iterable in a List. If the iterable argument is null,
+ * this method returns an empty list.
+ */
+ public static <E> List<E> iterableToList(@Nullable Iterable<E> iterable) {
+ if (iterable == null) {
+ return Collections.emptyList();
+ }
+
+ final ArrayList<E> list = new ArrayList<>();
+ iterable.iterator().forEachRemaining(list::add);
+ return list;
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
index 84c0a60..4b17b0d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
@@ -38,12 +38,11 @@ import org.apache.flink.runtime.source.event.ReaderRegistrationEvent;
import org.apache.flink.runtime.source.event.SourceEventWrapper;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.runtime.io.InputStatus;
import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput;
+import org.apache.flink.util.CollectionUtil;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@@ -66,19 +65,20 @@ public class SourceOperator<OUT, SplitT extends SourceSplit>
private final Source<OUT, SplitT, ?> source;
+ private final SimpleVersionedSerializer<SplitT> splitSerializer;
+
// Fields that will be setup at runtime.
private transient SourceReader<OUT, SplitT> sourceReader;
- private transient SimpleVersionedSerializer<SplitT> splitSerializer;
- private transient ListState<byte[]> readerState;
+ private transient ListState<SplitT> readerState;
private transient OperatorEventGateway operatorEventGateway;
public SourceOperator(Source<OUT, SplitT, ?> source) {
this.source = source;
+ this.splitSerializer = source.getSplitSerializer();
}
@Override
public void open() throws Exception {
- splitSerializer = source.getSplitSerializer();
// Create the source reader.
SourceReaderContext context = new SourceReaderContext() {
@Override
@@ -94,16 +94,11 @@ public class SourceOperator<OUT, SplitT extends SourceSplit>
sourceReader = source.createReader(context);
// restore the state if necessary.
- if (readerState.get() != null && readerState.get().iterator().hasNext()) {
- List<SplitT> splits = new ArrayList<>();
- for (byte[] splitBytes : readerState.get()) {
- SplitStateAndVersion stateWithVersion = SplitStateAndVersion.fromBytes(splitBytes);
- splits.add(splitSerializer.deserialize(
- stateWithVersion.getSerializerVersion(),
- stateWithVersion.getSplitState()));
- }
+ final List<SplitT> splits = CollectionUtil.iterableToList(readerState.get());
+ if (!splits.isEmpty()) {
sourceReader.addSplits(splits);
}
+
// Start the reader.
sourceReader.start();
// Register the reader to the coordinator.
@@ -128,15 +123,7 @@ public class SourceOperator<OUT, SplitT extends SourceSplit>
@Override
public void snapshotState(StateSnapshotContext context) throws Exception {
LOG.debug("Taking a snapshot for checkpoint {}", context.getCheckpointId());
- List<SplitT> splitStates = sourceReader.snapshotState();
- List<byte[]> state = new ArrayList<>();
- for (SplitT splitState : splitStates) {
- SplitStateAndVersion stateWithVersion = new SplitStateAndVersion(
- splitSerializer.getVersion(),
- splitSerializer.serialize(splitState));
- state.add(stateWithVersion.toBytes());
- }
- readerState.update(state);
+ readerState.update(sourceReader.snapshotState());
}
@Override
@@ -147,7 +134,8 @@ public class SourceOperator<OUT, SplitT extends SourceSplit>
@Override
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
- readerState = context.getOperatorStateStore().getListState(SPLITS_STATE_DESC);
+ final ListState<byte[]> rawState = context.getOperatorStateStore().getListState(SPLITS_STATE_DESC);
+ readerState = new SimpleVersionedListState<>(rawState, splitSerializer);
}
public void setOperatorEventGateway(OperatorEventGateway operatorEventGateway) {
@@ -178,43 +166,8 @@ public class SourceOperator<OUT, SplitT extends SourceSplit>
return sourceReader;
}
- // --------------- private class -----------------
-
- /**
- * Static container class. Package private for testing.
- */
@VisibleForTesting
- static class SplitStateAndVersion {
- private final int serializerVersion;
- private final byte[] splitState;
-
- SplitStateAndVersion(int serializerVersion, byte[] splitState) {
- this.serializerVersion = serializerVersion;
- this.splitState = splitState;
- }
-
- int getSerializerVersion() {
- return serializerVersion;
- }
-
- byte[] getSplitState() {
- return splitState;
- }
-
- byte[] toBytes() {
- // 4 Bytes - Serialization Version
- // N Bytes - Serialized Split State
- ByteBuffer buf = ByteBuffer.allocate(Integer.BYTES + Integer.BYTES + splitState.length);
- buf.putInt(serializerVersion);
- buf.put(splitState);
- return buf.array();
- }
-
- static SplitStateAndVersion fromBytes(byte[] bytes) {
- ByteBuffer buf = ByteBuffer.wrap(bytes);
- int version = buf.getInt();
- byte[] splitState = Arrays.copyOfRange(bytes, buf.position(), buf.limit());
- return new SplitStateAndVersion(version, splitState);
- }
+ ListState<SplitT> getReaderState() {
+ return readerState;
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java
index 65239c8..453b083 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.api.connector.source.mocks.MockSourceReader;
import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
import org.apache.flink.api.connector.source.mocks.MockSourceSplitSerializer;
import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.io.SimpleVersionedSerialization;
import org.apache.flink.runtime.operators.coordination.MockOperatorEventGateway;
import org.apache.flink.runtime.operators.coordination.OperatorEvent;
import org.apache.flink.runtime.operators.testutils.MockEnvironment;
@@ -41,11 +42,11 @@ import org.apache.flink.runtime.state.StateInitializationContextImpl;
import org.apache.flink.runtime.state.StateSnapshotContextSynchronousImpl;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.util.MockStreamingRuntimeContext;
+import org.apache.flink.util.CollectionUtil;
import org.junit.Before;
import org.junit.Test;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -134,17 +135,7 @@ public class SourceOperatorTest {
operator.snapshotState(new StateSnapshotContextSynchronousImpl(100L, 100L));
// Verify the splits in state.
- List<MockSourceSplit> splitsInState = new ArrayList<>();
- Iterable<byte[]> serializedSplits =
- stateContext.getOperatorStateStore().getListState(SourceOperator.SPLITS_STATE_DESC).get();
- for (byte[] serialized : serializedSplits) {
- MockSourceSplitSerializer serializer = new MockSourceSplitSerializer();
- SourceOperator.SplitStateAndVersion stateAndVersion =
- SourceOperator.SplitStateAndVersion.fromBytes(serialized);
- splitsInState.add(serializer.deserialize(
- stateAndVersion.getSerializerVersion(),
- stateAndVersion.getSplitState()));
- }
+ List<MockSourceSplit> splitsInState = CollectionUtil.iterableToList(operator.getReaderState().get());
assertEquals(Arrays.asList(MOCK_SPLIT, newSplit), splitsInState);
}
@@ -152,9 +143,8 @@ public class SourceOperatorTest {
private StateInitializationContext getStateContext() throws Exception {
// Create a mock split.
- byte[] serializedSplit = new MockSourceSplitSerializer().serialize(MOCK_SPLIT);
- byte[] serializedSplitWithVersion =
- new SourceOperator.SplitStateAndVersion(0, serializedSplit).toBytes();
+ byte[] serializedSplitWithVersion = SimpleVersionedSerialization
+ .writeVersionAndSerialize(new MockSourceSplitSerializer(), MOCK_SPLIT);
// Crate the state context.
OperatorStateStore operatorStateStore = createOperatorStateStore();