[FLINK-2916] [streaming] Expose operator and task information to Stat…

…eBackend

flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackend.java

@@ -18,96 +18,98 @@
 
 package org.apache.flink.runtime.state;
 
-import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataInputViewStreamWrapper;
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.runtime.execution.Environment;
 
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.Serializable;
 
 /**
  * A state backend defines how state is stored and snapshotted during checkpoints.
- * 
+ *
  * @param <Backend> The type of backend itself. This generic parameter is used to refer to the
  * type of backend when creating state backed by this backend.
  */
 public abstract class StateBackend<Backend extends StateBackend<Backend>> implements java.io.Serializable {
- 
+
  private static final long serialVersionUID = 4620413814639220247L;
- 
+
  // ------------------------------------------------------------------------
  // initialization and cleanup
  // ------------------------------------------------------------------------
- 
+
  /**
  * This method is called by the task upon deployment to initialize the state backend for
  * data for a specific job.
- * 
- * @param job The ID of the job for which the state backend instance checkpoints data.
+ *
+ * @param The {@link Environment} of the task that instantiated the state backend 
  * @throws Exception Overwritten versions of this method may throw exceptions, in which
  * case the job that uses the state backend is considered failed during
  * deployment.
  */
- public abstract void initializeForJob(JobID job) throws Exception;
+ public abstract void initializeForJob(Environment env) throws Exception;
 
  /**
  * Disposes all state associated with the current job.
- * 
+ *
  * @throws Exception Exceptions may occur during disposal of the state and should be forwarded.
  */
  public abstract void disposeAllStateForCurrentJob() throws Exception;
 
  /**
  * Closes the state backend, releasing all internal resources, but does not delete any persistent
  * checkpoint data.
- * 
+ *
  * @throws Exception Exceptions can be forwarded and will be logged by the system
  */
  public abstract void close() throws Exception;
- 
+
  // ------------------------------------------------------------------------
  // key/value state
  // ------------------------------------------------------------------------
 
  /**
  * Creates a key/value state backed by this state backend.
- * 
+ *
+ * @param operatorId Unique id for the operator creating the state
+ * @param stateName Name of the created state
  * @param keySerializer The serializer for the key.
  * @param valueSerializer The serializer for the value.
  * @param defaultValue The value that is returned when no other value has been associated with a key, yet.
  * @param <K> The type of the key.
  * @param <V> The type of the value.
- * 
+ *
  * @return A new key/value state backed by this backend.
- * 
+ *
  * @throws Exception Exceptions may occur during initialization of the state and should be forwarded.
  */
- public abstract <K, V> KvState<K, V, Backend> createKvState(
+ public abstract <K, V> KvState<K, V, Backend> createKvState(int operatorId, String stateName,
  TypeSerializer<K> keySerializer, TypeSerializer<V> valueSerializer,
  V defaultValue) throws Exception;
- 
- 
+
+
  // ------------------------------------------------------------------------
  // storing state for a checkpoint
  // ------------------------------------------------------------------------
 
  /**
  * Creates an output stream that writes into the state of the given checkpoint. When the stream
  * is closes, it returns a state handle that can retrieve the state back.
- * 
+ *
  * @param checkpointID The ID of the checkpoint.
  * @param timestamp The timestamp of the checkpoint.
  * @return An output stream that writes state for the given checkpoint.
- * 
+ *
  * @throws Exception Exceptions may occur while creating the stream and should be forwarded.
  */
  public abstract CheckpointStateOutputStream createCheckpointStateOutputStream(
  long checkpointID, long timestamp) throws Exception;
- 
+
  /**
  * Creates a {@link DataOutputView} stream that writes into the state of the given checkpoint.
  * When the stream is closes, it returns a state handle that can retrieve the state back.
@@ -125,20 +127,20 @@ public CheckpointStateOutputView createCheckpointStateOutputView(
 
  /**
  * Writes the given state into the checkpoint, and returns a handle that can retrieve the state back.
- * 
+ *
  * @param state The state to be checkpointed.
  * @param checkpointID The ID of the checkpoint.
  * @param timestamp The timestamp of the checkpoint.
  * @param <S> The type of the state.
- * 
+ *
  * @return A state handle that can retrieve the checkpoined state.
- * 
+ *
  * @throws Exception Exceptions may occur during serialization / storing the state and should be forwarded.
  */
  public abstract <S extends Serializable> StateHandle<S> checkpointStateSerializable(
  S state, long checkpointID, long timestamp) throws Exception;
- 
- 
+
+
  // ------------------------------------------------------------------------
  // Checkpoint state output stream
  // ------------------------------------------------------------------------
@@ -151,7 +153,7 @@ public static abstract class CheckpointStateOutputStream extends OutputStream {
  /**
  * Closes the stream and gets a state handle that can create an input stream
  * producing the data written to this stream.
- * 
+ *
  * @return A state handle that can create an input stream producing the data written to this stream.
  * @throws IOException Thrown, if the stream cannot be closed.
  */
@@ -162,9 +164,9 @@ public static abstract class CheckpointStateOutputStream extends OutputStream {
  * A dedicated DataOutputView stream that produces a {@code StateHandle<DataInputView>} when closed.
  */
  public static final class CheckpointStateOutputView extends DataOutputViewStreamWrapper {
- 
+
  private final CheckpointStateOutputStream out;
- 
+
  public CheckpointStateOutputView(CheckpointStateOutputStream out) {
  super(out);
  this.out = out;
@@ -193,7 +195,7 @@ public void close() throws IOException {
  private static final class DataInputViewHandle implements StateHandle<DataInputView> {
 
  private static final long serialVersionUID = 2891559813513532079L;
- 
+
  private final StreamStateHandle stream;
 
  private DataInputViewHandle(StreamStateHandle stream) {
@@ -202,7 +204,7 @@ private DataInputViewHandle(StreamStateHandle stream) {
 
  @Override
  public DataInputView getState(ClassLoader userCodeClassLoader) throws Exception {
- return new DataInputViewStreamWrapper(stream.getState(userCodeClassLoader)); 
+ return new DataInputViewStreamWrapper(stream.getState(userCodeClassLoader));
  }
 
  @Override