[FLINK-2707] [streaming] Set StateCheckpointer before default state v…

…alue
nltran · Sep 18, 2015 · 3e233a3 · 3e233a3
1 parent b234b0b
commit 3e233a3
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 2 deletions.
diff --git a/...-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamingRuntimeContext.java b/...-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamingRuntimeContext.java
@@ -104,8 +104,8 @@ public <S, C extends Serializable> OperatorState<S> getOperatorState(String name
  throw new RuntimeException("Cannot set default state to null.");
  }
  StreamOperatorState<S, C> state = (StreamOperatorState<S, C>) getState(name, partitioned);
- state.setDefaultState(defaultState);
  state.setCheckpointer(checkpointer);
+ state.setDefaultState(defaultState);
 
  return (OperatorState<S>) state;
  }

diff --git a/...reaming-core/src/test/java/org/apache/flink/streaming/api/state/StatefulOperatorTest.java b/...reaming-core/src/test/java/org/apache/flink/streaming/api/state/StatefulOperatorTest.java
@@ -37,6 +37,7 @@
 import org.apache.flink.api.common.accumulators.Accumulator;
 import org.apache.flink.api.common.functions.RichMapFunction;
 import org.apache.flink.api.common.state.OperatorState;
+import org.apache.flink.api.common.state.StateCheckpointer;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
@@ -278,7 +279,19 @@ public String map(Integer value) throws Exception {
 
  @Override
  public void open(Configuration conf) throws IOException {
- groupCounter = getRuntimeContext().getOperatorState("groupCounter", 0, true);
+ groupCounter = getRuntimeContext().getOperatorState("groupCounter", 0, true,
+ new StateCheckpointer<Integer, String>() {
+
+ @Override
+ public String snapshotState(Integer state, long checkpointId, long checkpointTimestamp) {
+ return state.toString();
+ }
+
+ @Override
+ public Integer restoreState(String stateSnapshot) {
+ return Integer.parseInt(stateSnapshot);
+ }
+ });
  }
 
  @SuppressWarnings("unchecked")