flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From srich...@apache.org
Subject flink git commit: [FLINK-6964] [checkpoint] Fix externalized incremental checkpoints for StandaloneCompletedCheckpointStore
Date Mon, 17 Jul 2017 12:51:32 GMT
Repository: flink
Updated Branches:
  refs/heads/release-1.3 1ed2ef438 -> 4b003eafd


[FLINK-6964] [checkpoint] Fix externalized incremental checkpoints for StandaloneCompletedCheckpointStore

(cherry picked from commit 8cff17f)


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4b003eaf
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4b003eaf
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4b003eaf

Branch: refs/heads/release-1.3
Commit: 4b003eafdeccf0179511f3cd4a6f3f5bc63cf921
Parents: 1ed2ef4
Author: Stefan Richter <s.richter@data-artisans.com>
Authored: Wed Jun 21 11:22:28 2017 +0200
Committer: Stefan Richter <s.richter@data-artisans.com>
Committed: Fri Jul 14 19:21:13 2017 +0200

----------------------------------------------------------------------
 .../state/RocksDBKeyedStateBackend.java         |  12 +-
 .../checkpoint/CheckpointCoordinator.java       |  34 ++-
 .../savepoint/SavepointV2Serializer.java        |   7 +-
 .../state/IncrementalKeyedStateHandle.java      |  21 +-
 .../runtime/state/SharedStateRegistry.java      |  45 +++-
 .../ZooKeeperCompletedCheckpointStoreTest.java  |  19 +-
 .../savepoint/CheckpointTestUtils.java          |   2 +-
 .../state/IncrementalKeyedStateHandleTest.java  |   3 +-
 .../runtime/testingUtils/TestingCluster.scala   |  47 +++-
 .../testingUtils/TestingJobManagerLike.scala    |  59 ++++-
 .../TestingJobManagerMessages.scala             |  39 +++-
 .../ExternalizedCheckpointITCase.java           | 227 +++++++++++++++++++
 12 files changed, 480 insertions(+), 35 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index 44046fb..a6b53ec 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -179,6 +179,9 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K>
{
 	/** The identifier of the last completed checkpoint */
 	private long lastCompletedCheckpointId = -1;
 
+	/** Unique ID of this backend. */
+	private UUID backendUID;
+
 	private static final String SST_FILE_SUFFIX = ".sst";
 
 	public RocksDBKeyedStateBackend(
@@ -227,6 +230,8 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K>
{
 		this.kvStateInformation = new HashMap<>();
 		this.restoredKvStateMetaInfos = new HashMap<>();
 		this.materializedSstFiles = new TreeMap<>();
+		this.backendUID = UUID.randomUUID();
+		LOG.debug("Setting initial keyed backend uid for operator {} to {}.", this.operatorIdentifier,
this.backendUID);
 	}
 
 	/**
@@ -897,7 +902,7 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K>
{
 			}
 
 			return new IncrementalKeyedStateHandle(
-				stateBackend.operatorIdentifier,
+				stateBackend.backendUID,
 				stateBackend.keyGroupRange,
 				checkpointId,
 				sstFiles,
@@ -1401,6 +1406,11 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K>
{
 						}
 					}
 				} else {
+					// pick up again the old backend id, so the we can reference existing state
+					stateBackend.backendUID = restoreStateHandle.getBackendIdentifier();
+
+					LOG.debug("Restoring keyed backend uid in operator {} from incremental snapshot to {}.",
+						stateBackend.operatorIdentifier, stateBackend.backendUID);
 
 					// create hard links in the instance directory
 					if (!stateBackend.instanceRocksDBPath.mkdirs()) {

http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index 5201d43..82933ac 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.checkpoint;
 
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.time.Time;
@@ -415,6 +416,36 @@ public class CheckpointCoordinator {
 		return triggerCheckpoint(timestamp, checkpointProperties, checkpointDirectory, isPeriodic).isSuccess();
 	}
 
+	/**
+	 * Test method to trigger a checkpoint/savepoint.
+	 *
+	 * @param timestamp The timestamp for the checkpoint.
+	 * @param options The checkpoint options.
+	 * @return A future to the completed checkpoint
+	 */
+	@VisibleForTesting
+	@Internal
+	public Future<CompletedCheckpoint> triggerCheckpoint(long timestamp, CheckpointOptions
options) throws Exception {
+		switch (options.getCheckpointType()) {
+			case SAVEPOINT:
+				return triggerSavepoint(timestamp, options.getTargetLocation());
+
+			case FULL_CHECKPOINT:
+				CheckpointTriggerResult triggerResult =
+					triggerCheckpoint(timestamp, checkpointProperties, checkpointDirectory, false);
+
+				if (triggerResult.isSuccess()) {
+					return triggerResult.getPendingCheckpoint().getCompletionFuture();
+				} else {
+					Throwable cause = new Exception("Failed to trigger checkpoint: " + triggerResult.getFailureReason().message());
+					return FlinkCompletableFuture.completedExceptionally(cause);
+				}
+
+			default:
+				throw new IllegalArgumentException("Unknown checkpoint type: " + options.getCheckpointType());
+		}
+	}
+
 	@VisibleForTesting
 	CheckpointTriggerResult triggerCheckpoint(
 			long timestamp,
@@ -1090,6 +1121,7 @@ public class CheckpointCoordinator {
 		CompletedCheckpoint savepoint = SavepointLoader.loadAndValidateSavepoint(
 				job, tasks, savepointPath, userClassLoader, allowNonRestored);
 
+		savepoint.registerSharedStatesAfterRestored(sharedStateRegistry);
 		completedCheckpointStore.addCheckpoint(savepoint);
 		
 		// Reset the checkpoint ID counter
@@ -1097,7 +1129,7 @@ public class CheckpointCoordinator {
 		checkpointIdCounter.setCount(nextCheckpointId);
 		
 		LOG.info("Reset the checkpoint ID to {}.", nextCheckpointId);
-		
+
 		return restoreLatestCheckpointedState(tasks, true, allowNonRestored);
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2Serializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2Serializer.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2Serializer.java
index da0022c..c8d695f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2Serializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2Serializer.java
@@ -45,6 +45,7 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 
 /**
  * (De)serializer for checkpoint metadata format version 2.
@@ -320,7 +321,7 @@ class SavepointV2Serializer implements SavepointSerializer<SavepointV2>
{
 			dos.writeByte(INCREMENTAL_KEY_GROUPS_HANDLE);
 
 			dos.writeLong(incrementalKeyedStateHandle.getCheckpointId());
-			dos.writeUTF(incrementalKeyedStateHandle.getOperatorIdentifier());
+			dos.writeUTF(String.valueOf(incrementalKeyedStateHandle.getBackendIdentifier()));
 			dos.writeInt(incrementalKeyedStateHandle.getKeyGroupRange().getStartKeyGroup());
 			dos.writeInt(incrementalKeyedStateHandle.getKeyGroupRange().getNumberOfKeyGroups());
 
@@ -380,7 +381,7 @@ class SavepointV2Serializer implements SavepointSerializer<SavepointV2>
{
 		} else if (INCREMENTAL_KEY_GROUPS_HANDLE == type) {
 
 			long checkpointId = dis.readLong();
-			String operatorId = dis.readUTF();
+			String backendId = dis.readUTF();
 			int startKeyGroup = dis.readInt();
 			int numKeyGroups = dis.readInt();
 			KeyGroupRange keyGroupRange =
@@ -391,7 +392,7 @@ class SavepointV2Serializer implements SavepointSerializer<SavepointV2>
{
 			Map<StateHandleID, StreamStateHandle> privateStates = deserializeStreamStateHandleMap(dis);
 
 			return new IncrementalKeyedStateHandle(
-				operatorId,
+				UUID.fromString(backendId),
 				keyGroupRange,
 				checkpointId,
 				sharedStates,

http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalKeyedStateHandle.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalKeyedStateHandle.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalKeyedStateHandle.java
index 770b5a9..0085890 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalKeyedStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/IncrementalKeyedStateHandle.java
@@ -20,10 +20,12 @@ package org.apache.flink.runtime.state;
 
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.util.Preconditions;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Map;
+import java.util.UUID;
 
 /**
  * The handle to states of an incremental snapshot.
@@ -57,9 +59,10 @@ public class IncrementalKeyedStateHandle implements KeyedStateHandle {
 	private static final long serialVersionUID = -8328808513197388231L;
 
 	/**
-	 * The operator instance identifier for this handle
+	 * UUID to identify the backend which created this state handle. This is in creating the
key for the
+	 * {@link SharedStateRegistry}.
 	 */
-	private final String operatorIdentifier;
+	private final UUID backendIdentifier;
 
 	/**
 	 * The key-group range covered by this state handle
@@ -97,14 +100,14 @@ public class IncrementalKeyedStateHandle implements KeyedStateHandle
{
 	private transient SharedStateRegistry sharedStateRegistry;
 
 	public IncrementalKeyedStateHandle(
-		String operatorIdentifier,
+		UUID backendIdentifier,
 		KeyGroupRange keyGroupRange,
 		long checkpointId,
 		Map<StateHandleID, StreamStateHandle> sharedState,
 		Map<StateHandleID, StreamStateHandle> privateState,
 		StreamStateHandle metaStateHandle) {
 
-		this.operatorIdentifier = Preconditions.checkNotNull(operatorIdentifier);
+		this.backendIdentifier = Preconditions.checkNotNull(backendIdentifier);
 		this.keyGroupRange = Preconditions.checkNotNull(keyGroupRange);
 		this.checkpointId = checkpointId;
 		this.sharedState = Preconditions.checkNotNull(sharedState);
@@ -134,8 +137,8 @@ public class IncrementalKeyedStateHandle implements KeyedStateHandle {
 		return metaStateHandle;
 	}
 
-	public String getOperatorIdentifier() {
-		return operatorIdentifier;
+	public UUID getBackendIdentifier() {
+		return backendIdentifier;
 	}
 
 	@Override
@@ -231,7 +234,7 @@ public class IncrementalKeyedStateHandle implements KeyedStateHandle {
 	 */
 	@VisibleForTesting
 	public SharedStateRegistryKey createSharedStateRegistryKeyFromFileName(StateHandleID shId)
{
-		return new SharedStateRegistryKey(operatorIdentifier + '-' + keyGroupRange, shId);
+		return new SharedStateRegistryKey(String.valueOf(backendIdentifier) + '-' + keyGroupRange,
shId);
 	}
 
 	/**
@@ -252,7 +255,7 @@ public class IncrementalKeyedStateHandle implements KeyedStateHandle {
 		if (getCheckpointId() != that.getCheckpointId()) {
 			return false;
 		}
-		if (!getOperatorIdentifier().equals(that.getOperatorIdentifier())) {
+		if (!getBackendIdentifier().equals(that.getBackendIdentifier())) {
 			return false;
 		}
 		if (!getKeyGroupRange().equals(that.getKeyGroupRange())) {
@@ -273,7 +276,7 @@ public class IncrementalKeyedStateHandle implements KeyedStateHandle {
 	@VisibleForTesting
 	@Override
 	public int hashCode() {
-		int result = getOperatorIdentifier().hashCode();
+		int result = getBackendIdentifier().hashCode();
 		result = 31 * result + getKeyGroupRange().hashCode();
 		result = 31 * result + (int) (getCheckpointId() ^ (getCheckpointId() >>> 32));
 		result = 31 * result + getSharedState().hashCode();

http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-runtime/src/main/java/org/apache/flink/runtime/state/SharedStateRegistry.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/SharedStateRegistry.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/SharedStateRegistry.java
index af9ac9d..e0ca873 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/SharedStateRegistry.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/SharedStateRegistry.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.state;
 
 import org.apache.flink.runtime.concurrent.Executors;
 import org.apache.flink.util.Preconditions;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -47,6 +48,7 @@ public class SharedStateRegistry {
 	/** Executor for async state deletion */
 	private final Executor asyncDisposalExecutor;
 
+	/** Default uses direct executor to delete unreferenced state */
 	public SharedStateRegistry() {
 		this(Executors.directExecutor());
 	}
@@ -83,11 +85,16 @@ public class SharedStateRegistry {
 			entry = registeredStates.get(registrationKey);
 
 			if (entry == null) {
+
+				// Additional check that should never fail, because only state handles that are not placeholders
should
+				// ever be inserted to the registry.
+				Preconditions.checkState(!isPlaceholder(state), "Attempt to reference unknown state:
" + registrationKey);
+
 				entry = new SharedStateRegistry.SharedStateEntry(state);
 				registeredStates.put(registrationKey, entry);
 			} else {
 				// delete if this is a real duplicate
-				if (!Objects.equals(state, entry.state)) {
+				if (!Objects.equals(state, entry.stateHandle)) {
 					scheduledStateDeletion = state;
 				}
 				entry.increaseReferenceCount();
@@ -95,6 +102,7 @@ public class SharedStateRegistry {
 		}
 
 		scheduleAsyncDelete(scheduledStateDeletion);
+		LOG.trace("Registered shared state {} under key {}.", entry, registrationKey);
 		return new Result(entry);
 	}
 
@@ -112,9 +120,10 @@ public class SharedStateRegistry {
 
 		final Result result;
 		final StreamStateHandle scheduledStateDeletion;
+		SharedStateRegistry.SharedStateEntry entry;
 
 		synchronized (registeredStates) {
-			SharedStateRegistry.SharedStateEntry entry = registeredStates.get(registrationKey);
+			entry = registeredStates.get(registrationKey);
 
 			Preconditions.checkState(entry != null,
 				"Cannot unregister a state that is not registered.");
@@ -124,7 +133,7 @@ public class SharedStateRegistry {
 			// Remove the state from the registry when it's not referenced any more.
 			if (entry.getReferenceCount() <= 0) {
 				registeredStates.remove(registrationKey);
-				scheduledStateDeletion = entry.getState();
+				scheduledStateDeletion = entry.getStateHandle();
 				result = new Result(null, 0);
 			} else {
 				scheduledStateDeletion = null;
@@ -132,6 +141,7 @@ public class SharedStateRegistry {
 			}
 		}
 
+		LOG.trace("Unregistered shared state {} under key {}.", entry, registrationKey);
 		scheduleAsyncDelete(scheduledStateDeletion);
 		return result;
 	}
@@ -142,6 +152,7 @@ public class SharedStateRegistry {
 	 * @param stateHandles The shared states to register.
 	 */
 	public void registerAll(Iterable<? extends CompositeStateHandle> stateHandles) {
+
 		if (stateHandles == null) {
 			return;
 		}
@@ -156,6 +167,8 @@ public class SharedStateRegistry {
 	private void scheduleAsyncDelete(StreamStateHandle streamStateHandle) {
 		// We do the small optimization to not issue discards for placeholders, which are NOPs.
 		if (streamStateHandle != null && !isPlaceholder(streamStateHandle)) {
+
+			LOG.trace("Scheduled delete of state handle {}.", streamStateHandle);
 			asyncDisposalExecutor.execute(
 				new SharedStateRegistry.AsyncDisposalRunnable(streamStateHandle));
 		}
@@ -171,18 +184,18 @@ public class SharedStateRegistry {
 	private static class SharedStateEntry {
 
 		/** The shared state handle */
-		private final StreamStateHandle state;
+		private final StreamStateHandle stateHandle;
 
 		/** The current reference count of the state handle */
 		private int referenceCount;
 
 		SharedStateEntry(StreamStateHandle value) {
-			this.state = value;
+			this.stateHandle = value;
 			this.referenceCount = 1;
 		}
 
-		StreamStateHandle getState() {
-			return state;
+		StreamStateHandle getStateHandle() {
+			return stateHandle;
 		}
 
 		int getReferenceCount() {
@@ -196,6 +209,14 @@ public class SharedStateRegistry {
 		void decreaseReferenceCount() {
 			--referenceCount;
 		}
+
+		@Override
+		public String toString() {
+			return "SharedStateEntry{" +
+				"stateHandle=" + stateHandle +
+				", referenceCount=" + referenceCount +
+				'}';
+		}
 	}
 
 	/**
@@ -210,7 +231,7 @@ public class SharedStateRegistry {
 		private final int referenceCount;
 
 		private Result(SharedStateEntry sharedStateEntry) {
-			this.reference = sharedStateEntry.getState();
+			this.reference = sharedStateEntry.getStateHandle();
 			this.referenceCount = sharedStateEntry.getReferenceCount();
 		}
 
@@ -228,6 +249,14 @@ public class SharedStateRegistry {
 		public int getReferenceCount() {
 			return referenceCount;
 		}
+
+		@Override
+		public String toString() {
+			return "Result{" +
+				"reference=" + reference +
+				", referenceCount=" + referenceCount +
+				'}';
+		}
 	}
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
index b5854dd..91bab85 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreTest.java
@@ -18,12 +18,6 @@
 
 package org.apache.flink.runtime.checkpoint;
 
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.api.BackgroundCallback;
-import org.apache.curator.framework.api.CuratorEvent;
-import org.apache.curator.framework.api.CuratorEventType;
-import org.apache.curator.framework.api.ErrorListenerPathable;
-import org.apache.curator.utils.EnsurePath;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.runtime.concurrent.Executors;
 import org.apache.flink.runtime.state.RetrievableStateHandle;
@@ -31,6 +25,13 @@ import org.apache.flink.runtime.state.SharedStateRegistry;
 import org.apache.flink.runtime.zookeeper.RetrievableStateStorageHelper;
 import org.apache.flink.runtime.zookeeper.ZooKeeperStateHandleStore;
 import org.apache.flink.util.TestLogger;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.api.BackgroundCallback;
+import org.apache.curator.framework.api.CuratorEvent;
+import org.apache.curator.framework.api.CuratorEventType;
+import org.apache.curator.framework.api.ErrorListenerPathable;
+import org.apache.curator.utils.EnsurePath;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mockito;
@@ -51,6 +52,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyCollection;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
@@ -160,9 +162,12 @@ public class ZooKeeperCompletedCheckpointStoreTest extends TestLogger
{
 			stateStorage,
 			Executors.directExecutor());
 
-		SharedStateRegistry sharedStateRegistry = new SharedStateRegistry();
+		SharedStateRegistry sharedStateRegistry = spy(new SharedStateRegistry());
 		zooKeeperCompletedCheckpointStore.recover(sharedStateRegistry);
 
+		verify(retrievableStateHandle1.retrieveState(), times(1)).registerSharedStatesAfterRestored(sharedStateRegistry);
+		verify(retrievableStateHandle2.retrieveState(), times(1)).registerSharedStatesAfterRestored(sharedStateRegistry);
+
 		CompletedCheckpoint latestCompletedCheckpoint = zooKeeperCompletedCheckpointStore.getLatestCheckpoint();
 
 		// check that we return the latest retrievable checkpoint

http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/CheckpointTestUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/CheckpointTestUtils.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/CheckpointTestUtils.java
index f985573..de1f599 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/CheckpointTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/CheckpointTestUtils.java
@@ -274,7 +274,7 @@ public class CheckpointTestUtils {
 
 	public static IncrementalKeyedStateHandle createDummyIncrementalKeyedStateHandle(Random
rnd) {
 		return new IncrementalKeyedStateHandle(
-			createRandomUUID(rnd).toString(),
+			createRandomUUID(rnd),
 			new KeyGroupRange(1, 1),
 			42L,
 			createRandomStateHandleMap(rnd),

http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-runtime/src/test/java/org/apache/flink/runtime/state/IncrementalKeyedStateHandleTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/IncrementalKeyedStateHandleTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/state/IncrementalKeyedStateHandleTest.java
index 2a6975a..c1b3ccd 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/IncrementalKeyedStateHandleTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/IncrementalKeyedStateHandleTest.java
@@ -23,6 +23,7 @@ import org.junit.Test;
 
 import java.util.Map;
 import java.util.Random;
+import java.util.UUID;
 
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -187,7 +188,7 @@ public class IncrementalKeyedStateHandleTest {
 
 	private static IncrementalKeyedStateHandle create(Random rnd) {
 		return new IncrementalKeyedStateHandle(
-			"test",
+			UUID.nameUUIDFromBytes("test".getBytes()),
 			KeyGroupRange.of(0, 0),
 			1L,
 			placeSpies(CheckpointTestUtils.createRandomStateHandleMap(rnd)),

http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
index 7980cdf..1b61c61 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
@@ -28,8 +28,8 @@ import akka.testkit.CallingThreadDispatcher
 import org.apache.flink.api.common.JobID
 import org.apache.flink.configuration.{ConfigConstants, Configuration}
 import org.apache.flink.runtime.akka.AkkaUtils
-import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory
 import org.apache.flink.runtime.checkpoint.savepoint.Savepoint
+import org.apache.flink.runtime.checkpoint.{CheckpointOptions, CheckpointRecoveryFactory}
 import org.apache.flink.runtime.clusterframework.FlinkResourceManager
 import org.apache.flink.runtime.clusterframework.types.ResourceIDRetrievable
 import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager
@@ -39,11 +39,12 @@ import org.apache.flink.runtime.instance.{ActorGateway, InstanceManager}
 import org.apache.flink.runtime.jobmanager.scheduler.Scheduler
 import org.apache.flink.runtime.jobmanager.{JobManager, MemoryArchivist, SubmittedJobGraphStore}
 import org.apache.flink.runtime.leaderelection.LeaderElectionService
+import org.apache.flink.runtime.messages.JobManagerMessages
 import org.apache.flink.runtime.messages.JobManagerMessages._
 import org.apache.flink.runtime.metrics.MetricRegistry
 import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster
 import org.apache.flink.runtime.taskmanager.TaskManager
-import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.ResponseSavepoint
+import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.{CheckpointRequest,
CheckpointRequestFailure, CheckpointRequestSuccess, ResponseSavepoint}
 import org.apache.flink.runtime.testingUtils.TestingMessages.Alive
 import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages.NotifyWhenRegisteredAtJobManager
 import org.apache.flink.runtime.testutils.TestingResourceManager
@@ -372,6 +373,48 @@ class TestingCluster(
       case _ => throw new IOException("Dispose savepoint failed")
     }
   }
+
+  @throws(classOf[IOException])
+  def requestCheckpoint(jobId: JobID, options : CheckpointOptions): String = {
+    val jobManagerGateway = getLeaderGateway(timeout)
+
+    // wait until the cluster is ready to take a checkpoint.
+    val allRunning = jobManagerGateway.ask(
+      TestingJobManagerMessages.WaitForAllVerticesToBeRunning(jobId), timeout)
+
+    Await.ready(allRunning, timeout)
+
+    // trigger checkpoint
+    val result = Await.result(
+      jobManagerGateway.ask(CheckpointRequest(jobId, options), timeout), timeout)
+
+    result match {
+      case success: CheckpointRequestSuccess => success.path
+      case fail: CheckpointRequestFailure => {
+        if (fail.cause.getMessage.contains("tasks not ready")) {
+          // retry if the tasks are not ready yet.
+          Thread.sleep(50)
+          requestCheckpoint(jobId, options)
+        } else {
+          throw new IOException(fail.cause)
+        }
+      }
+      case _ => throw new IllegalStateException("Trigger checkpoint failed")
+    }
+  }
+
+  @throws[Exception]
+  def cancelJob(jobId: JobID): Unit = {
+    if (getCurrentlyRunningJobsJava.contains(jobId)) {
+      val jobManagerGateway = getLeaderGateway(timeout)
+      val cancelFuture = jobManagerGateway.ask(new JobManagerMessages.CancelJob(jobId), timeout)
+      val result = Await.result(cancelFuture, timeout)
+      if (!result.isInstanceOf[JobManagerMessages.CancellationSuccess]) {
+        throw new Exception("Cancellation failed")
+      }
+    }
+    else throw new IllegalStateException("Job is not running")
+  }
  }
 
 object TestingCluster {

http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
index a3d31f5..3d3af95 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerLike.scala
@@ -22,14 +22,17 @@ import akka.actor.{ActorRef, Cancellable, Terminated}
 import akka.pattern.{ask, pipe}
 import org.apache.flink.api.common.JobID
 import org.apache.flink.runtime.FlinkActor
+import org.apache.flink.runtime.checkpoint.CheckpointOptions.CheckpointType
+import org.apache.flink.runtime.checkpoint.CompletedCheckpoint
 import org.apache.flink.runtime.checkpoint.savepoint.SavepointStore
+import org.apache.flink.runtime.concurrent.BiFunction
 import org.apache.flink.runtime.execution.ExecutionState
 import org.apache.flink.runtime.jobgraph.JobStatus
 import org.apache.flink.runtime.jobmanager.JobManager
 import org.apache.flink.runtime.jobmanager.slots.ActorTaskManagerGateway
 import org.apache.flink.runtime.messages.Acknowledge
 import org.apache.flink.runtime.messages.ExecutionGraphMessages.JobStatusChanged
-import org.apache.flink.runtime.messages.JobManagerMessages.{GrantLeadership, RegisterJobClient,
RequestClassloadingProps}
+import org.apache.flink.runtime.messages.JobManagerMessages._
 import org.apache.flink.runtime.messages.Messages.Disconnect
 import org.apache.flink.runtime.messages.RegistrationMessages.RegisterTaskManager
 import org.apache.flink.runtime.messages.TaskManagerMessages.Heartbeat
@@ -336,6 +339,60 @@ trait TestingJobManagerLike extends FlinkActor {
         }
       }
 
+    case CheckpointRequest(jobId, checkpointOptions) =>
+      currentJobs.get(jobId) match {
+        case Some((graph, _)) =>
+          val checkpointCoordinator = graph.getCheckpointCoordinator()
+
+          if (checkpointCoordinator != null) {
+            // Immutable copy for the future
+            val senderRef = sender()
+            try {
+              // Do this async, because checkpoint coordinator operations can
+              // contain blocking calls to the state backend or ZooKeeper.
+              val cpFuture = checkpointCoordinator.triggerCheckpoint(
+                System.currentTimeMillis(),
+                checkpointOptions)
+
+              cpFuture.handleAsync[Void](
+                new BiFunction[CompletedCheckpoint, Throwable, Void] {
+                  override def apply(success: CompletedCheckpoint, cause: Throwable): Void
= {
+                    if (success != null) {
+                      if (success.getExternalPointer == null &&
+                        CheckpointType.SAVEPOINT.equals(checkpointOptions.getCheckpointType))
{
+                        senderRef ! CheckpointRequestFailure(
+                          jobId, new Exception("Savepoint has not been persisted.")
+                        )
+                      } else {
+                        senderRef ! CheckpointRequestSuccess(
+                          jobId,
+                          success.getCheckpointID,
+                          success.getExternalPointer,
+                          success.getTimestamp)
+                      }
+                    } else {
+                      senderRef ! CheckpointRequestFailure(
+                        jobId, new Exception("Failed to complete checkpoint", cause))
+                    }
+                    null
+                  }
+                },
+                context.dispatcher)
+            } catch {
+              case e: Exception =>
+                senderRef ! CheckpointRequestFailure(jobId, new Exception(
+                  "Failed to trigger checkpoint", e))
+            }
+          } else {
+            sender() ! CheckpointRequestFailure(jobId, new IllegalStateException(
+              "Checkpointing disabled. You can enable it via the execution environment of
" +
+                "your job."))
+          }
+
+        case None =>
+          sender() ! CheckpointRequestFailure(jobId, new IllegalArgumentException("Unknown
job."))
+      }
+
     case NotifyWhenLeader =>
       if (leaderElectionService.hasLeadership) {
         sender() ! true

http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
index 48c2bfb..f79c124 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
@@ -23,10 +23,13 @@ import java.util.Map
 import akka.actor.ActorRef
 import org.apache.flink.api.common.JobID
 import org.apache.flink.api.common.accumulators.Accumulator
+import org.apache.flink.runtime.checkpoint.CheckpointOptions
 import org.apache.flink.runtime.checkpoint.savepoint.Savepoint
-import org.apache.flink.runtime.executiongraph.{AccessExecutionGraph, ExecutionAttemptID,
ExecutionGraph}
+import org.apache.flink.runtime.executiongraph.AccessExecutionGraph
 import org.apache.flink.runtime.instance.ActorGateway
 import org.apache.flink.runtime.jobgraph.JobStatus
+import org.apache.flink.runtime.messages.RequiresLeaderSessionID
+import org.apache.flink.runtime.messages.checkpoint.AbstractCheckpointMessage
 
 object TestingJobManagerMessages {
 
@@ -59,6 +62,40 @@ object TestingJobManagerMessages {
   case class TaskManagerTerminated(taskManager: ActorRef)
 
   /**
+    * Triggers a checkpoint for the specified job.
+    *
+    * This is not a subtype of [[AbstractCheckpointMessage]], because it is a
+    * control-flow message, which is *not* part of the checkpointing mechanism
+    * of triggering and acknowledging checkpoints.
+    *
+    * @param jobId The JobID of the job to trigger the savepoint for.
+    * @param options properties of the checkpoint
+    */
+  case class CheckpointRequest(
+    jobId: JobID,
+    options: CheckpointOptions) extends RequiresLeaderSessionID
+
+  /**
+    * Response after a successful checkpoint trigger containing the savepoint path.
+    *
+    * @param jobId The job ID for which the savepoint was triggered.
+    * @param path  The path of the savepoint.
+    */
+  case class CheckpointRequestSuccess(
+    jobId: JobID,
+    checkpointId: Long,
+    path: String,
+    triggerTime: Long)
+
+  /**
+    * Response after a failed checkpoint trigger containing the failure cause.
+    *
+    * @param jobId The job ID for which the savepoint was triggered.
+    * @param cause The cause of the failure.
+    */
+  case class CheckpointRequestFailure(jobId: JobID, cause: Throwable)
+
+  /**
    * Registers a listener to receive a message when accumulators changed.
    * The change must be explicitly triggered by the TestingTaskManager which can receive
an
    * [[org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages.AccumulatorsChanged]]

http://git-wip-us.apache.org/repos/asf/flink/blob/4b003eaf/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ExternalizedCheckpointITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ExternalizedCheckpointITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ExternalizedCheckpointITCase.java
new file mode 100644
index 0000000..c7696b4
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ExternalizedCheckpointITCase.java
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.checkpointing;
+
+import org.apache.flink.api.common.JobSubmissionResult;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
+import org.apache.flink.runtime.state.AbstractStateBackend;
+import org.apache.flink.runtime.state.filesystem.FsStateBackend;
+import org.apache.flink.runtime.state.filesystem.FsStateBackendFactory;
+import org.apache.flink.runtime.testingUtils.TestingCluster;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.test.state.ManualWindowSpeedITCase;
+
+import org.apache.curator.test.TestingServer;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+
+/**
+ * IT case for externalized checkpoints with {@link org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore}
+ * and {@link org.apache.flink.runtime.checkpoint.ZooKeeperCompletedCheckpointStore}.
+ *
+ * <p>This tests considers full and incremental checkpoints and was introduced to guard
against problems like FLINK-6964.
+ */
+public class ExternalizedCheckpointITCase {
+
+	private static final int PARALLELISM = 2;
+	private static final int NUM_TASK_MANAGERS = 2;
+	private static final int SLOTS_PER_TASK_MANAGER = 2;
+
+	@ClassRule
+	public static TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+	@Test
+	public void testExternalizedIncrementalRocksDBCheckpointsStandalone() throws Exception {
+		final File checkpointDir = temporaryFolder.newFolder();
+		testExternalizedCheckpoints(
+			checkpointDir,
+			null,
+			new RocksDBStateBackend(checkpointDir.toURI().toString(), true));
+	}
+
+	@Test
+	public void testExternalizedFullRocksDBCheckpointsStandalone() throws Exception {
+		final File checkpointDir = temporaryFolder.newFolder();
+		testExternalizedCheckpoints(
+			checkpointDir,
+			null,
+			new RocksDBStateBackend(checkpointDir.toURI().toString(), false));
+	}
+
+	@Test
+	public void testExternalizedFSCheckpointsStandalone() throws Exception {
+		final File checkpointDir = temporaryFolder.newFolder();
+		testExternalizedCheckpoints(
+			checkpointDir,
+			null,
+			new FsStateBackend(checkpointDir.toURI().toString(), true));
+
+	}
+
+	@Test
+	public void testExternalizedIncrementalRocksDBCheckpointsZookeeper() throws Exception {
+		TestingServer zkServer = new TestingServer();
+		zkServer.start();
+		try {
+			final File checkpointDir = temporaryFolder.newFolder();
+			testExternalizedCheckpoints(
+				checkpointDir,
+				zkServer.getConnectString(),
+				new RocksDBStateBackend(checkpointDir.toURI().toString(), true));
+		} finally {
+			zkServer.stop();
+		}
+	}
+
+	@Test
+	public void testExternalizedFullRocksDBCheckpointsZookeeper() throws Exception {
+		TestingServer zkServer = new TestingServer();
+		zkServer.start();
+		try {
+			final File checkpointDir = temporaryFolder.newFolder();
+			testExternalizedCheckpoints(
+				checkpointDir,
+				zkServer.getConnectString(),
+				new RocksDBStateBackend(checkpointDir.toURI().toString(), false));
+		} finally {
+			zkServer.stop();
+		}
+	}
+
+	@Test
+	public void testExternalizedFSCheckpointsZookeeper() throws Exception {
+		TestingServer zkServer = new TestingServer();
+		zkServer.start();
+		try {
+			final File checkpointDir = temporaryFolder.newFolder();
+			testExternalizedCheckpoints(
+				checkpointDir,
+				zkServer.getConnectString(),
+				new FsStateBackend(checkpointDir.toURI().toString(), true));
+		} finally {
+			zkServer.stop();
+		}
+	}
+
+	private void testExternalizedCheckpoints(
+		File checkpointDir,
+		String zooKeeperQuorum,
+		AbstractStateBackend backend) throws Exception {
+
+		final Configuration config = new Configuration();
+
+		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TASK_MANAGERS);
+		config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, SLOTS_PER_TASK_MANAGER);
+
+		final File savepointDir = temporaryFolder.newFolder();
+
+		config.setString(FsStateBackendFactory.CHECKPOINT_DIRECTORY_URI_CONF_KEY, checkpointDir.toURI().toString());
+		config.setString(ConfigConstants.SAVEPOINT_DIRECTORY_KEY, savepointDir.toURI().toString());
+		config.setString(ConfigConstants.CHECKPOINTS_DIRECTORY_KEY, checkpointDir.toURI().toString());
+
+		// ZooKeeper recovery mode?
+		if (zooKeeperQuorum != null) {
+			final File haDir = temporaryFolder.newFolder();
+			config.setString(HighAvailabilityOptions.HA_MODE, "ZOOKEEPER");
+			config.setString(HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM, zooKeeperQuorum);
+			config.setString(HighAvailabilityOptions.HA_STORAGE_PATH, haDir.toURI().toString());
+		}
+
+		TestingCluster cluster = new TestingCluster(config);
+		cluster.start();
+
+		String externalCheckpoint = null;
+
+		try {
+
+			// main test sequence:  start job -> eCP -> restore job -> eCP -> restore
job -> eCP
+			for (int i = 0; i < 3; ++i) {
+				final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+				env.setStateBackend(backend);
+				env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+
+				env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
+				env.setParallelism(PARALLELISM);
+
+				env.addSource(new ManualWindowSpeedITCase.InfiniteTupleSource(10_000))
+					.keyBy(0)
+					.timeWindow(Time.seconds(3))
+					.reduce(new ReduceFunction<Tuple2<String, Integer>>() {
+						private static final long serialVersionUID = 1L;
+
+						@Override
+						public Tuple2<String, Integer> reduce(
+							Tuple2<String, Integer> value1,
+							Tuple2<String, Integer> value2) throws Exception {
+							return Tuple2.of(value1.f0, value1.f1 + value2.f1);
+						}
+					})
+					.filter(new FilterFunction<Tuple2<String, Integer>>() {
+						private static final long serialVersionUID = 1L;
+
+						@Override
+						public boolean filter(Tuple2<String, Integer> value) throws Exception {
+							return value.f0.startsWith("Tuple 0");
+						}
+					});
+
+				StreamGraph streamGraph = env.getStreamGraph();
+				streamGraph.setJobName("Test");
+
+				JobGraph jobGraph = streamGraph.getJobGraph();
+
+				// recover from previous iteration?
+				if (externalCheckpoint != null) {
+					jobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(externalCheckpoint));
+				}
+
+				config.addAll(jobGraph.getJobConfiguration());
+				JobSubmissionResult submissionResult = cluster.submitJobDetached(jobGraph);
+
+				// let the job do some progress
+				Thread.sleep(200);
+
+				externalCheckpoint =
+					cluster.requestCheckpoint(submissionResult.getJobID(), CheckpointOptions.forFullCheckpoint());
+
+				cluster.cancelJob(submissionResult.getJobID());
+			}
+		} finally {
+			cluster.stop();
+			cluster.awaitTermination();
+		}
+	}
+}


Mime
View raw message