[FLINK-2986] Fix typo in KvState interface snapshot method

docs/internals/stream_checkpointing.md

@@ -52,7 +52,7 @@ Kafka exploits this ability.
 ## Checkpointing
 
 The central part of Flink's fault tolerance mechanism is drawing consistent snapshots of the distributed data stream and operator state.
-These shapshots act as consistent checkpoints to which the system can fall back in case of a failure. Flink's mechanism for drawing these
+These snapshots act as consistent checkpoints to which the system can fall back in case of a failure. Flink's mechanism for drawing these
 snapshots is described in "[Lightweight Asynchronous Snapshots for Distributed Dataflows](http://arxiv.org/abs/1506.08603)". It is inspired by 
 the standard [Chandy-Lamport algorithm](http://research.microsoft.com/en-us/um/people/lamport/pubs/chandy.pdf) for distributed snapshots and is 
 specifically tailored to Flink's execution model.

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

@@ -51,7 +51,7 @@ public interface KvState<K, V, Backend extends StateBackend<Backend>> extends Op
 	 * @throws Exception Exceptions during snapshotting the state should be forwarded, so the system
 	 *                   can react to failed snapshots.
 	 */
-	KvStateSnapshot<K, V, Backend> shapshot(long checkpointId, long timestamp) throws Exception;
+	KvStateSnapshot<K, V, Backend> snapshot(long checkpointId, long timestamp) throws Exception;
 
 	/**
 	 * Gets the number of key/value pairs currently stored in the state. Note that is a key

flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsHeapKvState.java

@@ -69,7 +69,7 @@ public class FsHeapKvState<K, V> extends AbstractHeapKvState<K, V, FsStateBacken
 
 	
 	@Override
-	public FsHeapKvStateSnapshot<K, V> shapshot(long checkpointId, long timestamp) throws Exception {
+	public FsHeapKvStateSnapshot<K, V> snapshot(long checkpointId, long timestamp) throws Exception {
 		// first, create an output stream to write to
 		try (FsStateBackend.FsCheckpointStateOutputStream out = 
 					backend.createCheckpointStateOutputStream(checkpointId, timestamp)) {

flink-runtime/src/main/java/org/apache/flink/runtime/state/memory/MemHeapKvState.java

@@ -42,7 +42,7 @@ public class MemHeapKvState<K, V> extends AbstractHeapKvState<K, V, MemoryStateB
 	}
 	
 	@Override
-	public MemoryHeapKvStateSnapshot<K, V> shapshot(long checkpointId, long timestamp) throws Exception {
+	public MemoryHeapKvStateSnapshot<K, V> snapshot(long checkpointId, long timestamp) throws Exception {
 		DataOutputSerializer ser = new DataOutputSerializer(Math.max(size() * 16, 16));
 		writeStateToOutputView(ser);
 		byte[] bytes = ser.getCopyOfBuffer();

flink-runtime/src/test/java/org/apache/flink/runtime/state/FileStateBackendTest.java

@@ -247,7 +247,7 @@ public class FileStateBackendTest {
 
 			// draw a snapshot
 			KvStateSnapshot<Integer, String, FsStateBackend> snapshot1 =
-					kv.shapshot(682375462378L, System.currentTimeMillis());
+					kv.snapshot(682375462378L, System.currentTimeMillis());
 
 			// make some more modifications
 			kv.setCurrentKey(1);
@@ -259,7 +259,7 @@ public class FileStateBackendTest {
 
 			// draw another snapshot
 			KvStateSnapshot<Integer, String, FsStateBackend> snapshot2 =
-					kv.shapshot(682375462379L, System.currentTimeMillis());
+					kv.snapshot(682375462379L, System.currentTimeMillis());
 
 			// validate the original state
 			assertEquals(3, kv.size());
@@ -325,7 +325,7 @@ public class FileStateBackendTest {
 			kv.update("2");
 
 			KvStateSnapshot<Integer, String, FsStateBackend> snapshot =
-					kv.shapshot(682375462378L, System.currentTimeMillis());
+					kv.snapshot(682375462378L, System.currentTimeMillis());
 
 
 			@SuppressWarnings("unchecked")

flink-runtime/src/test/java/org/apache/flink/runtime/state/MemoryStateBackendTest.java

@@ -166,7 +166,7 @@ public class MemoryStateBackendTest {
 			
 			// draw a snapshot
 			KvStateSnapshot<Integer, String, MemoryStateBackend> snapshot1 = 
-					kv.shapshot(682375462378L, System.currentTimeMillis());
+					kv.snapshot(682375462378L, System.currentTimeMillis());
 			
 			// make some more modifications
 			kv.setCurrentKey(1);
@@ -178,7 +178,7 @@ public class MemoryStateBackendTest {
 
 			// draw another snapshot
 			KvStateSnapshot<Integer, String, MemoryStateBackend> snapshot2 =
-					kv.shapshot(682375462379L, System.currentTimeMillis());
+					kv.snapshot(682375462379L, System.currentTimeMillis());
 			
 			// validate the original state
 			assertEquals(3, kv.size());
@@ -230,7 +230,7 @@ public class MemoryStateBackendTest {
 			kv.update("2");
 			
 			KvStateSnapshot<Integer, String, MemoryStateBackend> snapshot =
-					kv.shapshot(682375462378L, System.currentTimeMillis());
+					kv.snapshot(682375462378L, System.currentTimeMillis());
 
 
 			@SuppressWarnings("unchecked")

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java

@@ -161,7 +161,7 @@ public abstract class AbstractStreamOperator<OUT>
 			HashMap<String, KvStateSnapshot<?, ?, ?>> snapshots = new HashMap<>(keyValueStatesByName.size());
 			
 			for (Map.Entry<String, KvState<?, ?, ?>> entry : keyValueStatesByName.entrySet()) {
-				KvStateSnapshot<?, ?, ?> snapshot = entry.getValue().shapshot(checkpointId, timestamp);
+				KvStateSnapshot<?, ?, ?> snapshot = entry.getValue().snapshot(checkpointId, timestamp);
 				snapshots.put(entry.getKey(), snapshot);
 			}