You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2017/04/19 02:37:14 UTC
[1/2] flink git commit: [FLINK-6324] [DataStream] Refine
OperatorStateStore interface
Repository: flink
Updated Branches:
refs/heads/master 0182141d4 -> 2ef4900aa
[FLINK-6324] [DataStream] Refine OperatorStateStore interface
This commit refines the OperatorStateStore interface by,
- deprecating Java serialization shortcuts
- rename getOperatorState to getListState
The Java serialization shortcuts are deprecated because they were
previously introduced to provide a smoother migration path from older
savepoints. However, its usage should definitely be discouraged.
Renaming to getListState is a preparation of making the names of state
access methods contain information about both its redistribution pattern
on restore and the shape of its datastructure, since the combination of
these two is orthogonal. This convention will also provide a better
naming pattern for more state access methods in the future, for example
getUnionListState. If the method name does not contain its
redistribution pattern (e.g., getListState), then it simply implies the
default repartitioning scheme (SPLIT_DISTRIBUTE).
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a1aab642
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a1aab642
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a1aab642
Branch: refs/heads/master
Commit: a1aab64231f3177954404a1d160d8932d76b2826
Parents: 0182141
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Wed Apr 19 02:19:48 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Wed Apr 19 10:32:08 2017 +0800
----------------------------------------------------------------------
.../kafka/FlinkKafkaConsumerBaseTest.java | 6 +-
.../api/common/state/OperatorStateStore.java | 40 +++++-
.../ProcTimeUnboundedNonPartitionedOver.scala | 2 +-
.../state/DefaultOperatorStateBackend.java | 143 +++++++++++--------
.../runtime/state/OperatorStateBackendTest.java | 20 +--
.../ContinuousFileMonitoringFunction.java | 2 +-
.../functions/source/FromElementsFunction.java | 2 +-
.../source/StatefulSequenceSource.java | 2 +-
.../api/operators/async/AsyncWaitOperator.java | 4 +-
.../StreamOperatorSnapshotRestoreTest.java | 11 +-
.../WrappingFunctionSnapshotRestoreTest.java | 6 +-
.../runtime/tasks/OneInputStreamTaskTest.java | 4 +-
.../test/checkpointing/RescalingITCase.java | 7 +-
13 files changed, 158 insertions(+), 91 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
index 123c2be..4f5b283 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
@@ -109,7 +109,7 @@ public class FlinkKafkaConsumerBaseTest {
FlinkKafkaConsumerBase<String> consumer = getConsumer(fetcher, new LinkedMap(), false);
OperatorStateStore operatorStateStore = mock(OperatorStateStore.class);
TestingListState<Tuple2<KafkaTopicPartition, Long>> listState = new TestingListState<>();
- when(operatorStateStore.getOperatorState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
+ when(operatorStateStore.getListState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
consumer.snapshotState(new StateSnapshotContextSynchronousImpl(1, 1));
@@ -320,7 +320,7 @@ public class FlinkKafkaConsumerBaseTest {
OperatorStateStore operatorStateStore = mock(OperatorStateStore.class);
listState = new TestingListState<>();
- when(operatorStateStore.getOperatorState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
+ when(operatorStateStore.getListState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
// create 500 snapshots
for (int i = 100; i < 600; i++) {
@@ -444,7 +444,7 @@ public class FlinkKafkaConsumerBaseTest {
OperatorStateStore operatorStateStore = mock(OperatorStateStore.class);
listState = new TestingListState<>();
- when(operatorStateStore.getOperatorState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
+ when(operatorStateStore.getListState(Matchers.any(ListStateDescriptor.class))).thenReturn(listState);
// create 500 snapshots
for (int i = 100; i < 600; i++) {
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorStateStore.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorStateStore.java b/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorStateStore.java
index c1cdfe4..8be04fc 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorStateStore.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorStateStore.java
@@ -41,6 +41,34 @@ public interface OperatorStateStore {
* @return A list for all state partitions.
* @throws Exception
*/
+ <S> ListState<S> getListState(ListStateDescriptor<S> stateDescriptor) throws Exception;
+
+ /**
+ * Returns a set with the names of all currently registered states.
+ * @return set of names for all registered states.
+ */
+ Set<String> getRegisteredStateNames();
+
+ // -------------------------------------------------------------------------------------------
+ // Deprecated methods
+ // -------------------------------------------------------------------------------------------
+
+ /**
+ * Creates (or restores) a list state. Each state is registered under a unique name.
+ * The provided serializer is used to de/serialize the state in case of checkpointing (snapshot/restore).
+ *
+ * The items in the list are repartitionable by the system in case of changed operator parallelism.
+ *
+ * @param stateDescriptor The descriptor for this state, providing a name and serializer.
+ * @param <S> The generic type of the state
+ *
+ * @return A list for all state partitions.
+ * @throws Exception
+ *
+ * @deprecated since 1.3.0. This was deprecated as part of a refinement to the function names.
+ * Please use {@link #getListState(ListStateDescriptor)} instead.
+ */
+ @Deprecated
<S> ListState<S> getOperatorState(ListStateDescriptor<S> stateDescriptor) throws Exception;
/**
@@ -48,17 +76,15 @@ public interface OperatorStateStore {
* are repartitionable by the system in case of changed operator parallelism.
*
* <p>This is a simple convenience method. For more flexibility on how state serialization
- * should happen, use the {@link #getOperatorState(ListStateDescriptor)} method.
+ * should happen, use the {@link #getListState(ListStateDescriptor)} method.
*
* @param stateName The name of state to create
* @return A list state using Java serialization to serialize state objects.
* @throws Exception
+ *
+ * @deprecated since 1.3.0. Using Java serialization for persisting state is not encouraged.
+ * Please use {@link #getListState(ListStateDescriptor)} instead.
*/
+ @Deprecated
<T extends Serializable> ListState<T> getSerializableListState(String stateName) throws Exception;
-
- /**
- * Returns a set with the names of all currently registered states.
- * @return set of names for all registered states.
- */
- Set<String> getRegisteredStateNames();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/ProcTimeUnboundedNonPartitionedOver.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/ProcTimeUnboundedNonPartitionedOver.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/ProcTimeUnboundedNonPartitionedOver.scala
index 6b9800b..75209db 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/ProcTimeUnboundedNonPartitionedOver.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/ProcTimeUnboundedNonPartitionedOver.scala
@@ -91,6 +91,6 @@ class ProcTimeUnboundedNonPartitionedOver(
override def initializeState(context: FunctionInitializationContext): Unit = {
val accumulatorsDescriptor = new ListStateDescriptor[Row]("overState", aggregationStateType)
- state = context.getOperatorStateStore.getOperatorState(accumulatorsDescriptor)
+ state = context.getOperatorStateStore.getListState(accumulatorsDescriptor)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
index ca7cb48..42d4519 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
@@ -86,15 +86,13 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
registeredStates.clear();
}
- @SuppressWarnings("unchecked")
- @Override
- public <T extends Serializable> ListState<T> getSerializableListState(String stateName) throws Exception {
- return (ListState<T>) getOperatorState(new ListStateDescriptor<>(stateName, javaSerializer));
- }
+ // -------------------------------------------------------------------------------------------
+ // State access methods
+ // -------------------------------------------------------------------------------------------
@Override
- public <S> ListState<S> getOperatorState(ListStateDescriptor<S> stateDescriptor) throws IOException {
- return getOperatorState(stateDescriptor, OperatorStateHandle.Mode.SPLIT_DISTRIBUTE);
+ public <S> ListState<S> getListState(ListStateDescriptor<S> stateDescriptor) throws Exception {
+ return getListState(stateDescriptor, OperatorStateHandle.Mode.SPLIT_DISTRIBUTE);
}
@SuppressWarnings("unchecked")
@@ -103,64 +101,38 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
}
public <S> ListState<S> getBroadcastOperatorState(ListStateDescriptor<S> stateDescriptor) throws Exception {
- return getOperatorState(stateDescriptor, OperatorStateHandle.Mode.BROADCAST);
+ return getListState(stateDescriptor, OperatorStateHandle.Mode.BROADCAST);
}
- private <S> ListState<S> getOperatorState(
- ListStateDescriptor<S> stateDescriptor,
- OperatorStateHandle.Mode mode) throws IOException {
-
- Preconditions.checkNotNull(stateDescriptor);
-
- stateDescriptor.initializeSerializerUnlessSet(getExecutionConfig());
-
- String name = Preconditions.checkNotNull(stateDescriptor.getName());
- TypeSerializer<S> partitionStateSerializer = Preconditions.checkNotNull(stateDescriptor.getElementSerializer());
+ // -------------------------------------------------------------------------------------------
+ // Deprecated state access methods
+ // -------------------------------------------------------------------------------------------
- @SuppressWarnings("unchecked")
- PartitionableListState<S> partitionableListState = (PartitionableListState<S>) registeredStates.get(name);
-
- if (null == partitionableListState) {
-
- partitionableListState = new PartitionableListState<>(
- name,
- partitionStateSerializer,
- mode);
-
- registeredStates.put(name, partitionableListState);
- } else {
- Preconditions.checkState(
- partitionableListState.getAssignmentMode().equals(mode),
- "Incompatible assignment mode. Provided: " + mode + ", expected: " +
- partitionableListState.getAssignmentMode());
- Preconditions.checkState(
- stateDescriptor.getElementSerializer().
- canRestoreFrom(partitionableListState.getPartitionStateSerializer()),
- "Incompatible type serializers. Provided: " + stateDescriptor.getElementSerializer() +
- ", found: " + partitionableListState.getPartitionStateSerializer());
- }
-
- return partitionableListState;
+ /**
+ * @deprecated This was deprecated as part of a refinement to the function names.
+ * Please use {@link #getListState(ListStateDescriptor)} instead.
+ */
+ @Deprecated
+ @Override
+ public <S> ListState<S> getOperatorState(ListStateDescriptor<S> stateDescriptor) throws Exception {
+ return getListState(stateDescriptor);
}
- private static <S> void deserializeStateValues(
- PartitionableListState<S> stateListForName,
- FSDataInputStream in,
- OperatorStateHandle.StateMetaInfo metaInfo) throws IOException {
-
- if (null != metaInfo) {
- long[] offsets = metaInfo.getOffsets();
- if (null != offsets) {
- DataInputView div = new DataInputViewStreamWrapper(in);
- TypeSerializer<S> serializer = stateListForName.getPartitionStateSerializer();
- for (long offset : offsets) {
- in.seek(offset);
- stateListForName.add(serializer.deserialize(div));
- }
- }
- }
+ /**
+ * @deprecated Using Java serialization for persisting state is not encouraged.
+ * Please use {@link #getListState(ListStateDescriptor)} instead.
+ */
+ @SuppressWarnings("unchecked")
+ @Deprecated
+ @Override
+ public <T extends Serializable> ListState<T> getSerializableListState(String stateName) throws Exception {
+ return (ListState<T>) getListState(new ListStateDescriptor<>(stateName, javaSerializer));
}
+ // -------------------------------------------------------------------------------------------
+ // Snapshot and restore
+ // -------------------------------------------------------------------------------------------
+
@Override
public RunnableFuture<OperatorStateHandle> snapshot(
long checkpointId,
@@ -358,4 +330,59 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
return partitionOffsets;
}
}
+
+ private <S> ListState<S> getListState(
+ ListStateDescriptor<S> stateDescriptor,
+ OperatorStateHandle.Mode mode) throws IOException {
+
+ Preconditions.checkNotNull(stateDescriptor);
+
+ stateDescriptor.initializeSerializerUnlessSet(getExecutionConfig());
+
+ String name = Preconditions.checkNotNull(stateDescriptor.getName());
+ TypeSerializer<S> partitionStateSerializer = Preconditions.checkNotNull(stateDescriptor.getElementSerializer());
+
+ @SuppressWarnings("unchecked")
+ PartitionableListState<S> partitionableListState = (PartitionableListState<S>) registeredStates.get(name);
+
+ if (null == partitionableListState) {
+
+ partitionableListState = new PartitionableListState<>(
+ name,
+ partitionStateSerializer,
+ mode);
+
+ registeredStates.put(name, partitionableListState);
+ } else {
+ Preconditions.checkState(
+ partitionableListState.getAssignmentMode().equals(mode),
+ "Incompatible assignment mode. Provided: " + mode + ", expected: " +
+ partitionableListState.getAssignmentMode());
+ Preconditions.checkState(
+ stateDescriptor.getElementSerializer().
+ canRestoreFrom(partitionableListState.getPartitionStateSerializer()),
+ "Incompatible type serializers. Provided: " + stateDescriptor.getElementSerializer() +
+ ", found: " + partitionableListState.getPartitionStateSerializer());
+ }
+
+ return partitionableListState;
+ }
+
+ private static <S> void deserializeStateValues(
+ PartitionableListState<S> stateListForName,
+ FSDataInputStream in,
+ OperatorStateHandle.StateMetaInfo metaInfo) throws IOException {
+
+ if (null != metaInfo) {
+ long[] offsets = metaInfo.getOffsets();
+ if (null != offsets) {
+ DataInputView div = new DataInputViewStreamWrapper(in);
+ TypeSerializer<S> serializer = stateListForName.getPartitionStateSerializer();
+ for (long offset : offsets) {
+ in.seek(offset);
+ stateListForName.add(serializer.deserialize(div));
+ }
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
index bc446f3..5a072df 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
@@ -80,10 +80,10 @@ public class OperatorStateBackendTest {
ListStateDescriptor<File> stateDescriptor = new ListStateDescriptor<>("test", File.class);
ListStateDescriptor<String> stateDescriptor2 = new ListStateDescriptor<>("test2", String.class);
- ListState<File> listState = operatorStateBackend.getOperatorState(stateDescriptor);
+ ListState<File> listState = operatorStateBackend.getListState(stateDescriptor);
assertNotNull(listState);
- ListState<String> listState2 = operatorStateBackend.getOperatorState(stateDescriptor2);
+ ListState<String> listState2 = operatorStateBackend.getListState(stateDescriptor2);
assertNotNull(listState2);
assertEquals(2, operatorStateBackend.getRegisteredStateNames().size());
@@ -113,7 +113,7 @@ public class OperatorStateBackendTest {
ListStateDescriptor<Serializable> stateDescriptor1 = new ListStateDescriptor<>("test1", new JavaSerializer<>());
ListStateDescriptor<Serializable> stateDescriptor2 = new ListStateDescriptor<>("test2", new JavaSerializer<>());
ListStateDescriptor<Serializable> stateDescriptor3 = new ListStateDescriptor<>("test3", new JavaSerializer<>());
- ListState<Serializable> listState1 = operatorStateBackend.getOperatorState(stateDescriptor1);
+ ListState<Serializable> listState1 = operatorStateBackend.getListState(stateDescriptor1);
assertNotNull(listState1);
assertEquals(1, operatorStateBackend.getRegisteredStateNames().size());
Iterator<Serializable> it = listState1.get().iterator();
@@ -126,7 +126,7 @@ public class OperatorStateBackendTest {
assertEquals(4711, it.next());
assertTrue(!it.hasNext());
- ListState<Serializable> listState2 = operatorStateBackend.getOperatorState(stateDescriptor2);
+ ListState<Serializable> listState2 = operatorStateBackend.getListState(stateDescriptor2);
assertNotNull(listState2);
assertEquals(2, operatorStateBackend.getRegisteredStateNames().size());
assertTrue(!it.hasNext());
@@ -154,7 +154,7 @@ public class OperatorStateBackendTest {
assertEquals(123, it.next());
assertTrue(!it.hasNext());
- ListState<Serializable> listState1b = operatorStateBackend.getOperatorState(stateDescriptor1);
+ ListState<Serializable> listState1b = operatorStateBackend.getListState(stateDescriptor1);
assertNotNull(listState1b);
listState1b.add(123);
it = listState1b.get().iterator();
@@ -183,7 +183,7 @@ public class OperatorStateBackendTest {
}
try {
- operatorStateBackend.getOperatorState(stateDescriptor3);
+ operatorStateBackend.getListState(stateDescriptor3);
fail("Did not detect changed mode");
} catch (IllegalStateException ignored) {
@@ -217,8 +217,8 @@ public class OperatorStateBackendTest {
ListStateDescriptor<Serializable> stateDescriptor1 = new ListStateDescriptor<>("test1", new JavaSerializer<>());
ListStateDescriptor<Serializable> stateDescriptor2 = new ListStateDescriptor<>("test2", new JavaSerializer<>());
ListStateDescriptor<Serializable> stateDescriptor3 = new ListStateDescriptor<>("test3", new JavaSerializer<>());
- ListState<Serializable> listState1 = operatorStateBackend.getOperatorState(stateDescriptor1);
- ListState<Serializable> listState2 = operatorStateBackend.getOperatorState(stateDescriptor2);
+ ListState<Serializable> listState1 = operatorStateBackend.getListState(stateDescriptor1);
+ ListState<Serializable> listState2 = operatorStateBackend.getListState(stateDescriptor2);
ListState<Serializable> listState3 = operatorStateBackend.getBroadcastOperatorState(stateDescriptor3);
listState1.add(42);
@@ -251,8 +251,8 @@ public class OperatorStateBackendTest {
assertEquals(3, operatorStateBackend.getRegisteredStateNames().size());
- listState1 = operatorStateBackend.getOperatorState(stateDescriptor1);
- listState2 = operatorStateBackend.getOperatorState(stateDescriptor2);
+ listState1 = operatorStateBackend.getListState(stateDescriptor1);
+ listState2 = operatorStateBackend.getListState(stateDescriptor2);
listState3 = operatorStateBackend.getBroadcastOperatorState(stateDescriptor3);
assertEquals(3, operatorStateBackend.getRegisteredStateNames().size());
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileMonitoringFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileMonitoringFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileMonitoringFunction.java
index 589e285..a7539d1 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileMonitoringFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileMonitoringFunction.java
@@ -132,7 +132,7 @@ public class ContinuousFileMonitoringFunction<OUT>
Preconditions.checkState(this.checkpointedState == null,
"The " + getClass().getSimpleName() + " has already been initialized.");
- this.checkpointedState = context.getOperatorStateStore().getOperatorState(
+ this.checkpointedState = context.getOperatorStateStore().getListState(
new ListStateDescriptor<>(
"file-monitoring-state",
LongSerializer.INSTANCE
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
index e3a9d54..e294cae 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
@@ -104,7 +104,7 @@ public class FromElementsFunction<T> implements SourceFunction<T>, CheckpointedF
Preconditions.checkState(this.checkpointedState == null,
"The " + getClass().getSimpleName() + " has already been initialized.");
- this.checkpointedState = context.getOperatorStateStore().getOperatorState(
+ this.checkpointedState = context.getOperatorStateStore().getListState(
new ListStateDescriptor<>(
"from-elements-state",
IntSerializer.INSTANCE
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/StatefulSequenceSource.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/StatefulSequenceSource.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/StatefulSequenceSource.java
index bdb12f3..fb1c94d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/StatefulSequenceSource.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/StatefulSequenceSource.java
@@ -71,7 +71,7 @@ public class StatefulSequenceSource extends RichParallelSourceFunction<Long> imp
Preconditions.checkState(this.checkpointedState == null,
"The " + getClass().getSimpleName() + " has already been initialized.");
- this.checkpointedState = context.getOperatorStateStore().getOperatorState(
+ this.checkpointedState = context.getOperatorStateStore().getListState(
new ListStateDescriptor<>(
"stateful-sequence-source-state",
LongSerializer.INSTANCE
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
index 4cf79b1..ff507e3 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
@@ -243,7 +243,7 @@ public class AsyncWaitOperator<IN, OUT>
super.snapshotState(context);
ListState<StreamElement> partitionableState =
- getOperatorStateBackend().getOperatorState(new ListStateDescriptor<>(STATE_NAME, inStreamElementSerializer));
+ getOperatorStateBackend().getListState(new ListStateDescriptor<>(STATE_NAME, inStreamElementSerializer));
partitionableState.clear();
Collection<StreamElementQueueEntry<?>> values = queue.values();
@@ -269,7 +269,7 @@ public class AsyncWaitOperator<IN, OUT>
public void initializeState(StateInitializationContext context) throws Exception {
recoveredStreamElements = context
.getOperatorStateStore()
- .getOperatorState(new ListStateDescriptor<>(STATE_NAME, inStreamElementSerializer));
+ .getListState(new ListStateDescriptor<>(STATE_NAME, inStreamElementSerializer));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamOperatorSnapshotRestoreTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamOperatorSnapshotRestoreTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamOperatorSnapshotRestoreTest.java
index 08fbcbe..8a5c997 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamOperatorSnapshotRestoreTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamOperatorSnapshotRestoreTest.java
@@ -19,9 +19,11 @@
package org.apache.flink.streaming.api.operators;
import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
@@ -169,8 +171,13 @@ public class StreamOperatorSnapshotRestoreTest {
Assert.assertEquals(verifyRestore, context.isRestored());
- keyedState = context.getKeyedStateStore().getState(new ValueStateDescriptor<>("managed-keyed", Integer.class, 0));
- opState = context.getOperatorStateStore().getSerializableListState("managed-op-state");
+ keyedState = context
+ .getKeyedStateStore()
+ .getState(new ValueStateDescriptor<>("managed-keyed", Integer.class, 0));
+
+ opState = context
+ .getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("managed-op-state", IntSerializer.INSTANCE));
if (context.isRestored()) {
// check restored raw keyed state
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/WrappingFunctionSnapshotRestoreTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/WrappingFunctionSnapshotRestoreTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/WrappingFunctionSnapshotRestoreTest.java
index ab4258f..b675cc5 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/WrappingFunctionSnapshotRestoreTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/WrappingFunctionSnapshotRestoreTest.java
@@ -21,6 +21,8 @@ package org.apache.flink.streaming.api.operators;
import org.apache.flink.api.common.functions.AbstractRichFunction;
import org.apache.flink.api.common.functions.MapFunction;
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.java.operators.translation.WrappingFunction;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
@@ -139,7 +141,9 @@ public class WrappingFunctionSnapshotRestoreTest {
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
- serializableListState = context.getOperatorStateStore().getSerializableListState("test-state");
+ serializableListState = context
+ .getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("test-state", IntSerializer.INSTANCE));
if (context.isRestored()) {
Iterator<Integer> integers = serializableListState.get().iterator();
int act = integers.next();
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
index 4e8bfd8..90f5619 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
@@ -721,7 +721,7 @@ public class OneInputStreamTaskTest extends TestLogger {
public void open() throws Exception {
super.open();
- ListState<Integer> partitionableState = getOperatorStateBackend().getOperatorState(TEST_DESCRIPTOR);
+ ListState<Integer> partitionableState = getOperatorStateBackend().getListState(TEST_DESCRIPTOR);
if (numberSnapshotCalls == 0) {
for (Integer v : partitionableState.get()) {
@@ -742,7 +742,7 @@ public class OneInputStreamTaskTest extends TestLogger {
@Override
public void snapshotState(StateSnapshotContext context) throws Exception {
ListState<Integer> partitionableState =
- getOperatorStateBackend().getOperatorState(TEST_DESCRIPTOR);
+ getOperatorStateBackend().getListState(TEST_DESCRIPTOR);
partitionableState.clear();
partitionableState.add(42);
http://git-wip-us.apache.org/repos/asf/flink/blob/a1aab642/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
index 875d0ed..fa05e1d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
@@ -23,8 +23,10 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.ConfigConstants;
@@ -996,8 +998,9 @@ public class RescalingITCase extends TestLogger {
this.counterPartitions =
operatorStateStore.getBroadcastSerializableListState("counter_partitions");
} else {
- this.counterPartitions =
- context.getOperatorStateStore().getSerializableListState("counter_partitions");
+ this.counterPartitions = context
+ .getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("counter_partitions", IntSerializer.INSTANCE));
}
if (context.isRestored()) {
[2/2] flink git commit: [FLINK-5991] [DataStream] Expose Union
ListState for operator state
Posted by tz...@apache.org.
[FLINK-5991] [DataStream] Expose Union ListState for operator state
This commit exposes the union list state scheme for managed operator state.
The actual functionality was already previously added to the
`DefaultOperatorStateBackend`, so this change simply exposes the feature
through the `OperatorStateStore` interface.
This commit also updates the documentation for managed operator state so
that it covers the new union list state scheme. It strengthens the
difference between keyed and non-keyed state data structures by
emphasizing the semantic differences in the state access method Javadocs.
This closes #3508.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2ef4900a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2ef4900a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2ef4900a
Branch: refs/heads/master
Commit: 2ef4900aa279e75844a9f8536cfe007c2542187d
Parents: a1aab64
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Wed Apr 19 02:53:31 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Wed Apr 19 10:33:29 2017 +0800
----------------------------------------------------------------------
docs/dev/stream/state.md | 109 ++++++++++++-------
.../api/common/state/OperatorStateStore.java | 36 +++++-
.../state/DefaultOperatorStateBackend.java | 8 +-
.../runtime/state/OperatorStateBackendTest.java | 21 ++--
.../test/checkpointing/RescalingITCase.java | 9 +-
5 files changed, 122 insertions(+), 61 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2ef4900a/docs/dev/stream/state.md
----------------------------------------------------------------------
diff --git a/docs/dev/stream/state.md b/docs/dev/stream/state.md
index 40522e1..276842d 100644
--- a/docs/dev/stream/state.md
+++ b/docs/dev/stream/state.md
@@ -64,16 +64,12 @@ for one or more Key Groups.
With *Operator State* (or *non-keyed state*), each operator state is
bound to one parallel operator instance.
-The Kafka source connector is a good motivating example for the use of Operator State
-in Flink. Each parallel instance of this Kafka consumer maintains a map
+The [Kafka Connector](../connectors/kafka.html) is a good motivating example for the use of Operator State
+in Flink. Each parallel instance of the Kafka consumer maintains a map
of topic partitions and offsets as its Operator State.
The Operator State interfaces support redistributing state among
-parallel operator instances when the parallelism is changed. There can be different schemes for doing this redistribution; the following are currently defined:
-
- - **List-style redistribution:** Each operator returns a List of state elements. The whole state is logically a concatenation of
- all lists. On restore/redistribution, the list is evenly divided into as many sublists as there are parallel operators.
- Each operator gets a sublist, which can be empty, or contain one or more elements.
+parallel operator instances when the parallelism is changed. There can be different schemes for doing this redistribution.
## Raw and Managed State
@@ -233,32 +229,13 @@ val counts: DataStream[(String, Int)] = stream
## Using Managed Operator State
-A stateful function can implement either the more general `CheckpointedFunction`
+To use managed operator state, a stateful function can implement either the more general `CheckpointedFunction`
interface, or the `ListCheckpointed<T extends Serializable>` interface.
-In both cases, the non-keyed state is expected to be a `List` of *serializable* objects, independent from each other,
-thus eligible for redistribution upon rescaling. In other words, these objects are the finest granularity at which
-non-keyed state can be repartitioned. As an example, if with parallelism 1 the checkpointed state of the `BufferingSink`
-contains elements `(test1, 2)` and `(test2, 2)`, when increasing the parallelism to 2, `(test1, 2)` may end up in task 0,
-while `(test2, 2)` will go to task 1.
-
-##### ListCheckpointed
+#### CheckpointedFunction
-The `ListCheckpointed` interface requires the implementation of two methods:
-
-{% highlight java %}
-List<T> snapshotState(long checkpointId, long timestamp) throws Exception;
-
-void restoreState(List<T> state) throws Exception;
-{% endhighlight %}
-
-On `snapshotState()` the operator should return a list of objects to checkpoint and
-`restoreState` has to handle such a list upon recovery. If the state is not re-partitionable, you can always
-return a `Collections.singletonList(MY_STATE)` in the `snapshotState()`.
-
-##### CheckpointedFunction
-
-The `CheckpointedFunction` interface also requires the implementation of two methods:
+The `CheckpointedFunction` interface provides access to non-keyed state with different
+redistribution schemes. It requires the implementation of two methods:
{% highlight java %}
void snapshotState(FunctionSnapshotContext context) throws Exception;
@@ -266,12 +243,30 @@ void snapshotState(FunctionSnapshotContext context) throws Exception;
void initializeState(FunctionInitializationContext context) throws Exception;
{% endhighlight %}
-Whenever a checkpoint has to be performed `snapshotState()` is called. The counterpart, `initializeState()`, is called every time the user-defined function is initialized, be that when the function is first initialized
-or be that when actually recovering from an earlier checkpoint. Given this, `initializeState()` is not
+Whenever a checkpoint has to be performed, `snapshotState()` is called. The counterpart, `initializeState()`,
+is called every time the user-defined function is initialized, be that when the function is first initialized
+or be that when the function is actually recovering from an earlier checkpoint. Given this, `initializeState()` is not
only the place where different types of state are initialized, but also where state recovery logic is included.
-This is an example of a function that uses `CheckpointedFunction`, a stateful `SinkFunction` that
-uses state to buffer elements before sending them to the outside world:
+Currently, list-style managed operator state is supported. The state
+is expected to be a `List` of *serializable* objects, independent from each other,
+thus eligible for redistribution upon rescaling. In other words, these objects are the finest granularity at which
+non-keyed state can be redistributed. Depending on the state accessing method,
+the following redistribution schemes are defined:
+
+ - **Even-split redistribution:** Each operator returns a List of state elements. The whole state is logically a concatenation of
+ all lists. On restore/redistribution, the list is evenly divided into as many sublists as there are parallel operators.
+ Each operator gets a sublist, which can be empty, or contain one or more elements.
+ As an example, if with parallelism 1 the checkpointed state of an operator
+ contains elements `element1` and `element2`, when increasing the parallelism to 2, `element1` may end up in operator instance 0,
+ while `element2` will go to operator instance 1.
+
+ - **Union redistribution:** Each operator returns a List of state elements. The whole state is logically a concatenation of
+ all lists. On restore/redistribution, each operator gets the complete list of state elements.
+
+Below is an example of a stateful `SinkFunction` that uses `CheckpointedFunction`
+to buffer elements before sending them to the outside world. It demonstrates
+the basic even-split redistribution list state:
{% highlight java %}
public class BufferingSink
@@ -311,8 +306,13 @@ public class BufferingSink
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
- checkpointedState = context.getOperatorStateStore().
- getSerializableListState("buffered-elements");
+ ListStateDescriptor<Tuple2<String, Integer>> descriptor =
+ new ListStateDescriptor<>(
+ "buffered-elements",
+ TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}),
+ Tuple2.of(0L, 0L));
+
+ checkpointedState = context.getOperatorStateStore().getListState(descriptor);
if (context.isRestored()) {
for (Tuple2<String, Integer> element : checkpointedState.get()) {
@@ -329,12 +329,29 @@ public class BufferingSink
}
{% endhighlight %}
-
The `initializeState` method takes as argument a `FunctionInitializationContext`. This is used to initialize
the non-keyed state "containers". These are a container of type `ListState` where the non-keyed state objects
are going to be stored upon checkpointing.
-`this.checkpointedState = context.getOperatorStateStore().getSerializableListState("buffered-elements");`
+Note how the state is initialized, similar to keyed state,
+with a `StateDescriptor` that contains the state name and information
+about the type of the value that the state holds:
+
+{% highlight java %}
+ListStateDescriptor<Tuple2<String, Integer>> descriptor =
+ new ListStateDescriptor<>(
+ "buffered-elements",
+ TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}),
+ Tuple2.of(0L, 0L));
+
+checkpointedState = context.getOperatorStateStore().getListState(descriptor);
+{% endhighlight %}
+
+The naming convention of the state access methods contain its redistribution
+pattern followed by its state structure. For example, to use list state with the
+union redistribution scheme on restore, access the state by using `getUnionListState(descriptor)`.
+If the method name does not contain the redistribution pattern, *e.g.* `getListState(descriptor)`,
+it simply implies that the basic even-split redistribution scheme will be used.
After initializing the container, we use the `isRestored()` method of the context to check if we are
recovering after a failure. If this is `true`, *i.e.* we are recovering, the restore logic is applied.
@@ -346,6 +363,22 @@ of all objects included by the previous checkpoint, and is then filled with the
As a side note, the keyed state can also be initialized in the `initializeState()` method. This can be done
using the provided `FunctionInitializationContext`.
+#### ListCheckpointed
+
+The `ListCheckpointed` interface is a more limited variant of `CheckpointedFunction`,
+which only supports list-style state with even-split redistribution scheme on restore.
+It also requires the implementation of two methods:
+
+{% highlight java %}
+List<T> snapshotState(long checkpointId, long timestamp) throws Exception;
+
+void restoreState(List<T> state) throws Exception;
+{% endhighlight %}
+
+On `snapshotState()` the operator should return a list of objects to checkpoint and
+`restoreState` has to handle such a list upon recovery. If the state is not re-partitionable, you can always
+return a `Collections.singletonList(MY_STATE)` in the `snapshotState()`.
+
### Stateful Source Functions
Stateful sources require a bit more care as opposed to other operators.
http://git-wip-us.apache.org/repos/asf/flink/blob/2ef4900a/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorStateStore.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorStateStore.java b/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorStateStore.java
index 8be04fc..bf22041 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorStateStore.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorStateStore.java
@@ -33,7 +33,16 @@ public interface OperatorStateStore {
* Creates (or restores) a list state. Each state is registered under a unique name.
* The provided serializer is used to de/serialize the state in case of checkpointing (snapshot/restore).
*
- * The items in the list are repartitionable by the system in case of changed operator parallelism.
+ * <p>Note the semantic differences between an operator list state and a keyed list state
+ * (see {@link KeyedStateStore#getListState(ListStateDescriptor)}). Under the context of operator state,
+ * the list is a collection of state items that are independent from each other and eligible for redistribution
+ * across operator instances in case of changed operator parallelism. In other words, these state items are
+ * the finest granularity at which non-keyed state can be redistributed, and should not be correlated with
+ * each other.
+ *
+ * <p>The redistribution scheme of this list state upon operator rescaling is a round-robin pattern, such that
+ * the logical whole state (a concatenation of all the lists of state elements previously managed by each operator
+ * before the restore) is evenly divided into as many sublists as there are parallel operators.
*
* @param stateDescriptor The descriptor for this state, providing a name and serializer.
* @param <S> The generic type of the state
@@ -44,7 +53,32 @@ public interface OperatorStateStore {
<S> ListState<S> getListState(ListStateDescriptor<S> stateDescriptor) throws Exception;
/**
+ * Creates (or restores) a list state. Each state is registered under a unique name.
+ * The provided serializer is used to de/serialize the state in case of checkpointing (snapshot/restore).
+ *
+ * <p>Note the semantic differences between an operator list state and a keyed list state
+ * (see {@link KeyedStateStore#getListState(ListStateDescriptor)}). Under the context of operator state,
+ * the list is a collection of state items that are independent from each other and eligible for redistribution
+ * across operator instances in case of changed operator parallelism. In other words, these state items are
+ * the finest granularity at which non-keyed state can be redistributed, and should not be correlated with
+ * each other.
+ *
+ * <p>The redistribution scheme of this list state upon operator rescaling is a broadcast pattern, such that
+ * the logical whole state (a concatenation of all the lists of state elements previously managed by each operator
+ * before the restore) is restored to all parallel operators so that each of them will get the union of all state
+ * items before the restore.
+ *
+ * @param stateDescriptor The descriptor for this state, providing a name and serializer.
+ * @param <S> The generic type of the state
+ *
+ * @return A list for all state partitions.
+ * @throws Exception
+ */
+ <S> ListState<S> getUnionListState(ListStateDescriptor<S> stateDescriptor) throws Exception;
+
+ /**
* Returns a set with the names of all currently registered states.
+ *
* @return set of names for all registered states.
*/
Set<String> getRegisteredStateNames();
http://git-wip-us.apache.org/repos/asf/flink/blob/2ef4900a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
index 42d4519..eb3ba01 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java
@@ -95,12 +95,8 @@ public class DefaultOperatorStateBackend implements OperatorStateBackend {
return getListState(stateDescriptor, OperatorStateHandle.Mode.SPLIT_DISTRIBUTE);
}
- @SuppressWarnings("unchecked")
- public <T extends Serializable> ListState<T> getBroadcastSerializableListState(String stateName) throws Exception {
- return (ListState<T>) getBroadcastOperatorState(new ListStateDescriptor<>(stateName, javaSerializer));
- }
-
- public <S> ListState<S> getBroadcastOperatorState(ListStateDescriptor<S> stateDescriptor) throws Exception {
+ @Override
+ public <S> ListState<S> getUnionListState(ListStateDescriptor<S> stateDescriptor) throws Exception {
return getListState(stateDescriptor, OperatorStateHandle.Mode.BROADCAST);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/2ef4900a/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
index 5a072df..13a6307 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/OperatorStateBackendTest.java
@@ -55,7 +55,7 @@ public class OperatorStateBackendTest {
public void testCreateOnAbstractStateBackend() throws Exception {
// we use the memory state backend as a subclass of the AbstractStateBackend
final AbstractStateBackend abstractStateBackend = new MemoryStateBackend();
- OperatorStateBackend operatorStateBackend = abstractStateBackend.createOperatorStateBackend(
+ final OperatorStateBackend operatorStateBackend = abstractStateBackend.createOperatorStateBackend(
createMockEnvironment(), "test-operator");
assertNotNull(operatorStateBackend);
@@ -75,7 +75,7 @@ public class OperatorStateBackendTest {
final ExecutionConfig cfg = new ExecutionConfig();
cfg.registerTypeWithKryoSerializer(registeredType, com.esotericsoftware.kryo.serializers.JavaSerializer.class);
- final DefaultOperatorStateBackend operatorStateBackend = new DefaultOperatorStateBackend(classLoader, cfg);
+ final OperatorStateBackend operatorStateBackend = new DefaultOperatorStateBackend(classLoader, cfg);
ListStateDescriptor<File> stateDescriptor = new ListStateDescriptor<>("test", File.class);
ListStateDescriptor<String> stateDescriptor2 = new ListStateDescriptor<>("test2", String.class);
@@ -107,7 +107,7 @@ public class OperatorStateBackendTest {
@Test
public void testRegisterStates() throws Exception {
- final DefaultOperatorStateBackend operatorStateBackend =
+ final OperatorStateBackend operatorStateBackend =
new DefaultOperatorStateBackend(classLoader, new ExecutionConfig());
ListStateDescriptor<Serializable> stateDescriptor1 = new ListStateDescriptor<>("test1", new JavaSerializer<>());
@@ -140,7 +140,7 @@ public class OperatorStateBackendTest {
assertEquals(23, it.next());
assertTrue(!it.hasNext());
- ListState<Serializable> listState3 = operatorStateBackend.getBroadcastOperatorState(stateDescriptor3);
+ ListState<Serializable> listState3 = operatorStateBackend.getUnionListState(stateDescriptor3);
assertNotNull(listState3);
assertEquals(3, operatorStateBackend.getRegisteredStateNames().size());
assertTrue(!it.hasNext());
@@ -176,7 +176,7 @@ public class OperatorStateBackendTest {
assertTrue(!it.hasNext());
try {
- operatorStateBackend.getBroadcastOperatorState(stateDescriptor2);
+ operatorStateBackend.getUnionListState(stateDescriptor2);
fail("Did not detect changed mode");
} catch (IllegalStateException ignored) {
@@ -194,7 +194,7 @@ public class OperatorStateBackendTest {
public void testSnapshotEmpty() throws Exception {
final AbstractStateBackend abstractStateBackend = new MemoryStateBackend(4096);
- final DefaultOperatorStateBackend operatorStateBackend = (DefaultOperatorStateBackend)
+ final OperatorStateBackend operatorStateBackend =
abstractStateBackend.createOperatorStateBackend(createMockEnvironment(), "testOperator");
CheckpointStreamFactory streamFactory =
@@ -211,7 +211,7 @@ public class OperatorStateBackendTest {
public void testSnapshotRestore() throws Exception {
AbstractStateBackend abstractStateBackend = new MemoryStateBackend(4096);
- DefaultOperatorStateBackend operatorStateBackend = (DefaultOperatorStateBackend)
+ OperatorStateBackend operatorStateBackend =
abstractStateBackend.createOperatorStateBackend(createMockEnvironment(), "test-op-name");
ListStateDescriptor<Serializable> stateDescriptor1 = new ListStateDescriptor<>("test1", new JavaSerializer<>());
@@ -219,7 +219,7 @@ public class OperatorStateBackendTest {
ListStateDescriptor<Serializable> stateDescriptor3 = new ListStateDescriptor<>("test3", new JavaSerializer<>());
ListState<Serializable> listState1 = operatorStateBackend.getListState(stateDescriptor1);
ListState<Serializable> listState2 = operatorStateBackend.getListState(stateDescriptor2);
- ListState<Serializable> listState3 = operatorStateBackend.getBroadcastOperatorState(stateDescriptor3);
+ ListState<Serializable> listState3 = operatorStateBackend.getUnionListState(stateDescriptor3);
listState1.add(42);
listState1.add(4711);
@@ -242,8 +242,7 @@ public class OperatorStateBackendTest {
operatorStateBackend.close();
operatorStateBackend.dispose();
- //TODO this is temporarily casted to test already functionality that we do not yet expose through public API
- operatorStateBackend = (DefaultOperatorStateBackend) abstractStateBackend.createOperatorStateBackend(
+ operatorStateBackend = abstractStateBackend.createOperatorStateBackend(
createMockEnvironment(),
"testOperator");
@@ -253,7 +252,7 @@ public class OperatorStateBackendTest {
listState1 = operatorStateBackend.getListState(stateDescriptor1);
listState2 = operatorStateBackend.getListState(stateDescriptor2);
- listState3 = operatorStateBackend.getBroadcastOperatorState(stateDescriptor3);
+ listState3 = operatorStateBackend.getUnionListState(stateDescriptor3);
assertEquals(3, operatorStateBackend.getRegisteredStateNames().size());
http://git-wip-us.apache.org/repos/asf/flink/blob/2ef4900a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
index fa05e1d..88dd1dd 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
@@ -145,7 +145,7 @@ public class RescalingITCase extends TestLogger {
}
/**
- * Tests that a a job with purely keyed state can be restarted from a savepoint
+ * Tests that a job with purely keyed state can be restarted from a savepoint
* with a different parallelism.
*/
public void testSavepointRescalingKeyedState(boolean scaleOut, boolean deriveMaxParallelism) throws Exception {
@@ -993,10 +993,9 @@ public class RescalingITCase extends TestLogger {
public void initializeState(FunctionInitializationContext context) throws Exception {
if (broadcast) {
- //TODO this is temporarily casted to test already functionality that we do not yet expose through public API
- DefaultOperatorStateBackend operatorStateStore = (DefaultOperatorStateBackend) context.getOperatorStateStore();
- this.counterPartitions =
- operatorStateStore.getBroadcastSerializableListState("counter_partitions");
+ this.counterPartitions = context
+ .getOperatorStateStore()
+ .getUnionListState(new ListStateDescriptor<>("counter_partitions", IntSerializer.INSTANCE));
} else {
this.counterPartitions = context
.getOperatorStateStore()