flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [2/2] flink git commit: [FLINK-5063] [checkpointing] Discard state handles of declined or expired state handles
Date Wed, 16 Nov 2016 23:36:44 GMT
[FLINK-5063] [checkpointing] Discard state handles of declined or expired state handles

Whenever the checkpoint coordinator receives an acknowledge checkpoint message which belongs
to the job maintained by the checkpoint coordinator, it should either record the state handles
for later processing or discard to free the resources. The latter case can happen if a
checkpoint has been expired and late acknowledge checkpoint messages arrive. Furthremore,
it
can happen if a Task sent a decline checkpoint message while other Tasks where still drawing
a checkpoint. This PR changes the behaviour such that state handles belonging to the job of
the checkpoint coordinator are discarded if they could not be added to the PendingCheckpoint.

This closes #2813


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

Branch: refs/heads/release-1.1
Commit: 4daf3bbc1e0251e1e84d799421dae9e3fa2363fc
Parents: ad3e674
Author: Till Rohrmann <trohrmann@apache.org>
Authored: Mon Nov 14 18:33:55 2016 +0100
Committer: Stephan Ewen <sewen@apache.org>
Committed: Wed Nov 16 21:29:32 2016 +0100

----------------------------------------------------------------------
 .../checkpoint/CheckpointCoordinator.java       |  88 ++++++++----
 .../runtime/checkpoint/PendingCheckpoint.java   | 103 ++++++++------
 .../checkpoint/CheckpointCoordinatorTest.java   | 134 ++++++++++++++++++-
 3 files changed, 256 insertions(+), 69 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/4daf3bbc/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 8661ddc..5c517b2 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
@@ -674,42 +674,65 @@ public class CheckpointCoordinator {
 			if (checkpoint != null && !checkpoint.isDiscarded()) {
 				isPendingCheckpoint = true;
 
-				if (checkpoint.acknowledgeTask(
-					message.getTaskExecutionId(),
-					message.getState(),
-					message.getStateSize(),
-					null)) { // TODO: Give KV-state to the acknowledgeTask method
-					if (checkpoint.isFullyAcknowledged()) {
-						completed = checkpoint.finalizeCheckpoint();
+				switch (checkpoint.acknowledgeTask(message.getTaskExecutionId(), message.getState(),
message.getStateSize(), null)) {
+					case SUCCESS:
+						// TODO: Give KV-state to the acknowledgeTask method
+						if (checkpoint.isFullyAcknowledged()) {
+							completed = checkpoint.finalizeCheckpoint();
 
-						completedCheckpointStore.addCheckpoint(completed);
+							completedCheckpointStore.addCheckpoint(completed);
 
-						LOG.info("Completed checkpoint " + checkpointId + " (in " +
+							LOG.info("Completed checkpoint " + checkpointId + " (in " +
 								completed.getDuration() + " ms)");
 
-						if (LOG.isDebugEnabled()) {
-							StringBuilder builder = new StringBuilder();
-							for (Map.Entry<JobVertexID, TaskState> entry: completed.getTaskStates().entrySet())
{
-								builder.append("JobVertexID: ").append(entry.getKey()).append(" {").append(entry.getValue()).append("}");
-							}
+							if (LOG.isDebugEnabled()) {
+								StringBuilder builder = new StringBuilder();
+								for (Map.Entry<JobVertexID, TaskState> entry : completed.getTaskStates().entrySet())
{
+									builder.append("JobVertexID: ").append(entry.getKey()).append(" {").append(entry.getValue()).append("}");
+								}
 
-							LOG.debug(builder.toString());
-						}
+								LOG.debug(builder.toString());
+							}
 
-						pendingCheckpoints.remove(checkpointId);
-						rememberRecentCheckpointId(checkpointId);
+							pendingCheckpoints.remove(checkpointId);
+							rememberRecentCheckpointId(checkpointId);
 
-						dropSubsumedCheckpoints(completed.getTimestamp());
+							dropSubsumedCheckpoints(completed.getTimestamp());
 
-						onFullyAcknowledgedCheckpoint(completed);
+							onFullyAcknowledgedCheckpoint(completed);
 
-						triggerQueuedRequests();
-					}
-				}
-				else {
-					// checkpoint did not accept message
-					LOG.error("Received duplicate or invalid acknowledge message for checkpoint " + checkpointId
-							+ " , task " + message.getTaskExecutionId());
+							triggerQueuedRequests();
+						}
+						break;
+					case DUPLICATE:
+						LOG.debug("Received a duplicate acknowledge message for checkpoint {}, task {}, job
{}.",
+							message.getCheckpointId(), message.getTaskExecutionId(), message.getJob());
+						break;
+					case UNKNOWN:
+						LOG.warn("Could not acknowledge the checkpoint {} for task {} of job {}, " +
+								"because the task's execution attempt id was unknown. Discarding " +
+								"the state handle to avoid lingering state.", message.getCheckpointId(),
+							message.getTaskExecutionId(), message.getJob());
+
+						try {
+							message.getState().deserializeValue(userClassLoader).discardState();
+						} catch (Exception e) {
+							LOG.warn("Could not properly discard state for checkpoint {} of task {} of job {}.",
+								message.getCheckpointId(), message.getTaskExecutionId(), message.getJob(), e);
+						}
+						break;
+					case DISCARDED:
+						LOG.warn("Could not acknowledge the checkpoint {} for task {} of job {}, " +
+							"because the pending checkpoint had been discarded. Discarding the " +
+								"state handle tp avoid lingering state.",
+							message.getCheckpointId(), message.getTaskExecutionId(), message.getJob());
+
+						try {
+							message.getState().deserializeValue(userClassLoader).discardState();
+						} catch (Exception e) {
+							LOG.warn("Could not properly discard state for checkpoint {} of task {} of job {}.",
+								message.getCheckpointId(), message.getTaskExecutionId(), message.getJob(), e);
+						}
 				}
 			}
 			else if (checkpoint != null) {
@@ -721,11 +744,20 @@ public class CheckpointCoordinator {
 				// message is for an unknown checkpoint, or comes too late (checkpoint disposed)
 				if (recentPendingCheckpoints.contains(checkpointId)) {
 					isPendingCheckpoint = true;
-					LOG.warn("Received late message for now expired checkpoint attempt " + checkpointId);
+					LOG.warn("Received late message for now expired checkpoint attempt {}.", checkpointId);
 				}
 				else {
+					LOG.debug("Received message for an unknown checkpoint {}.", checkpointId);
 					isPendingCheckpoint = false;
 				}
+
+				try {
+					// try to discard the state so that we don't have lingering state lying around
+					message.getState().deserializeValue(userClassLoader).discardState();
+				} catch (Exception e) {
+					LOG.warn("Could not properly discard state for checkpoint {} of task {} of job {}.",
+						message.getCheckpointId(), message.getTaskExecutionId(), message.getJob(), e);
+				}
 			}
 		}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/4daf3bbc/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
index cd1f6c4..e81bb09 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
@@ -27,6 +27,8 @@ import org.apache.flink.runtime.executiongraph.ExecutionVertex;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 import org.apache.flink.runtime.state.StateHandle;
 import org.apache.flink.util.SerializedValue;
+import java.util.HashSet;
+import java.util.Set;
 
 /**
  * A pending checkpoint is a checkpoint that has been started, but has not been
@@ -50,6 +52,9 @@ public class PendingCheckpoint {
 
 	private final Map<ExecutionAttemptID, ExecutionVertex> notYetAcknowledgedTasks;
 	
+	/** Set of acknowledged tasks */
+	private final Set<ExecutionAttemptID> acknowledgedTasks;
+
 	private int numAcknowledgedTasks;
 	
 	private boolean discarded;
@@ -72,6 +77,8 @@ public class PendingCheckpoint {
 		
 		this.notYetAcknowledgedTasks = verticesToConfirm;
 		this.taskStates = new HashMap<>();
+
+		acknowledgedTasks = new HashSet<>(verticesToConfirm.size());
 	}
 	
 	// --------------------------------------------------------------------------------------------
@@ -131,65 +138,82 @@ public class PendingCheckpoint {
 		}
 	}
 	
-	public boolean acknowledgeTask(
-			ExecutionAttemptID attemptID,
+	public TaskAcknowledgeResult acknowledgeTask(
+			ExecutionAttemptID executionAttemptId,
 			SerializedValue<StateHandle<?>> state,
 			long stateSize,
 			Map<Integer, SerializedValue<StateHandle<?>>> kvState) {
 
 		synchronized (lock) {
 			if (discarded) {
-				return false;
+				return TaskAcknowledgeResult.DISCARDED;
 			}
 			
-			ExecutionVertex vertex = notYetAcknowledgedTasks.remove(attemptID);
-			if (vertex != null) {
-				if (state != null || kvState != null) {
+			final ExecutionVertex vertex = notYetAcknowledgedTasks.remove(executionAttemptId);
+
+			if (vertex == null) {
+				if (acknowledgedTasks.contains(executionAttemptId)) {
+					return TaskAcknowledgeResult.DUPLICATE;
+				} else {
+					return TaskAcknowledgeResult.UNKNOWN;
+				}
+			} else {
+				acknowledgedTasks.add(executionAttemptId);
+			}
 
-					JobVertexID jobVertexID = vertex.getJobvertexId();
 
-					TaskState taskState;
+			if (state != null || kvState != null) {
+				JobVertexID jobVertexID = vertex.getJobvertexId();
 
-					if (taskStates.containsKey(jobVertexID)) {
-						taskState = taskStates.get(jobVertexID);
-					} else {
-						taskState = new TaskState(jobVertexID, vertex.getTotalNumberOfParallelSubtasks());
-						taskStates.put(jobVertexID, taskState);
-					}
+				TaskState taskState;
+
+				if (taskStates.containsKey(jobVertexID)) {
+					taskState = taskStates.get(jobVertexID);
+				} else {
+					taskState = new TaskState(jobVertexID, vertex.getTotalNumberOfParallelSubtasks());
+					taskStates.put(jobVertexID, taskState);
+				}
 
-					long timestamp = System.currentTimeMillis() - checkpointTimestamp;
+				long timestamp = System.currentTimeMillis() - checkpointTimestamp;
+
+				if (state != null) {
+					taskState.putState(
+						vertex.getParallelSubtaskIndex(),
+						new SubtaskState(
+							state,
+							stateSize,
+							timestamp
+						)
+					);
+				}
 
-					if (state != null) {
-						taskState.putState(
-							vertex.getParallelSubtaskIndex(),
-							new SubtaskState(
-								state,
-								stateSize,
+				if (kvState != null) {
+					for (Map.Entry<Integer, SerializedValue<StateHandle<?>>> entry : kvState.entrySet())
{
+						taskState.putKvState(
+							entry.getKey(),
+							new KeyGroupState(
+								entry.getValue(),
+								0L,
 								timestamp
-							)
-						);
-					}
+							));
 
-					if (kvState != null) {
-						for (Map.Entry<Integer, SerializedValue<StateHandle<?>>> entry :
kvState.entrySet()) {
-							taskState.putKvState(
-								entry.getKey(),
-								new KeyGroupState(
-									entry.getValue(),
-									0L,
-									timestamp
-								));
-						}
 					}
 				}
-				numAcknowledgedTasks++;
-				return true;
-			}
-			else {
-				return false;
 			}
+			numAcknowledgedTasks++;
+			return TaskAcknowledgeResult.SUCCESS;
 		}
 	}
+
+	/**
+	 * Result of the {@link PendingCheckpoint#acknowledgedTasks} method.
+	 */
+	public enum TaskAcknowledgeResult {
+		SUCCESS, // successful acknowledge of the task
+		DUPLICATE, // acknowledge message is a duplicate
+		UNKNOWN, // unknown task acknowledged
+		DISCARDED // pending checkpoint has been discarded
+	}
 	
 	/**
 	 * Discards the pending checkpoint, releasing all held resources.
@@ -211,6 +235,7 @@ public class PendingCheckpoint {
 			} finally {
 				taskStates.clear();
 				notYetAcknowledgedTasks.clear();
+				acknowledgedTasks.clear();
 			}
 		}
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/4daf3bbc/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
index f3f988a..f05b5d2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
@@ -30,6 +30,8 @@ import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
 import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
 import org.apache.flink.runtime.messages.checkpoint.NotifyCheckpointComplete;
 import org.apache.flink.runtime.messages.checkpoint.TriggerCheckpoint;
+import org.apache.flink.runtime.state.StateHandle;
+import org.apache.flink.util.SerializedValue;
 import org.junit.Test;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
@@ -49,8 +51,10 @@ import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.eq;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -954,6 +958,118 @@ public class CheckpointCoordinatorTest {
 		}
 	}
 
+	/**
+	 * Tests that late acknowledge checkpoint messages are properly cleaned up. Furthermore
it tests
+	 * that unknown checkpoint messages for the same job a are cleaned up as well. In contrast
+	 * checkpointing messages from other jobs should not be touched. A late acknowledge
+	 * message is an acknowledge message which arrives after the checkpoint has been declined.
+	 *
+	 * @throws Exception
+	 */
+	@Test
+	public void testStateCleanupForLateOrUnknownMessages() throws Exception {
+		final JobID jobId = new JobID();
+
+		final ExecutionAttemptID triggerAttemptId = new ExecutionAttemptID();
+		final ExecutionVertex triggerVertex = mockExecutionVertex(triggerAttemptId);
+
+		final ExecutionAttemptID ackAttemptId1 = new ExecutionAttemptID();
+		final ExecutionVertex ackVertex1 = mockExecutionVertex(ackAttemptId1);
+
+		final ExecutionAttemptID ackAttemptId2 = new ExecutionAttemptID();
+		final ExecutionVertex ackVertex2 = mockExecutionVertex(ackAttemptId2);
+
+		final long timestamp = 1L;
+
+		CheckpointCoordinator coord = new CheckpointCoordinator(
+			jobId,
+			20000L,
+			20000L,
+			42,
+			new ExecutionVertex[] { triggerVertex },
+			new ExecutionVertex[] {triggerVertex, ackVertex1, ackVertex2},
+			new ExecutionVertex[0],
+			cl,
+			new StandaloneCheckpointIDCounter(),
+			new StandaloneCompletedCheckpointStore(2, cl),
+			RecoveryMode.STANDALONE);
+
+		assertTrue(coord.triggerCheckpoint(timestamp));
+
+		assertEquals(1, coord.getNumberOfPendingCheckpoints());
+
+		PendingCheckpoint pendingCheckpoint = coord.getPendingCheckpoints().values().iterator().next();
+
+		long checkpointId = pendingCheckpoint.getCheckpointId();
+
+		StateHandle<?> triggerSubtaskState = mock(StateHandle.class);
+		SerializedValue<StateHandle<?>> triggerSerializedValue = mock(SerializedValue.class);
+		doReturn(triggerSubtaskState).when(triggerSerializedValue).deserializeValue(any(ClassLoader.class));
+
+		// acknowledge the first trigger vertex
+		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jobId, triggerAttemptId, checkpointId,
triggerSerializedValue, 1L));
+
+		StateHandle<?> unknownSubtaskState = mock(StateHandle.class);
+		SerializedValue<StateHandle<?>> unknownSerializedValue = mock(SerializedValue.class);
+		doReturn(unknownSubtaskState).when(unknownSerializedValue).deserializeValue(any(ClassLoader.class));
+
+		// receive an acknowledge message for an unknown vertex
+		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jobId, new ExecutionAttemptID(),
checkpointId, unknownSerializedValue, 1L));
+
+		// we should discard acknowledge messages from an unknown vertex belonging to our job
+		verify(unknownSubtaskState, times(1)).discardState();
+
+		StateHandle<?> differentJobSubtaskState = mock(StateHandle.class);
+		SerializedValue<StateHandle<?>> differentJobSerializedValue = mock(SerializedValue.class);
+		doReturn(differentJobSubtaskState).when(differentJobSerializedValue).deserializeValue(any(ClassLoader.class));
+
+		// receive an acknowledge message from an unknown job
+		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(new JobID(), new ExecutionAttemptID(),
checkpointId, differentJobSerializedValue, 1L));
+
+		// we should not interfere with different jobs
+		verify(differentJobSubtaskState, never()).discardState();
+
+		// duplicate acknowledge message for the trigger vertex
+		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jobId, triggerAttemptId, checkpointId,
triggerSerializedValue, 1L));
+
+		// duplicate acknowledge messages for a known vertex should not trigger discarding the
state
+		verify(triggerSubtaskState, never()).discardState();
+
+		// let the checkpoint fail at the first ack vertex
+		coord.receiveDeclineMessage(new DeclineCheckpoint(jobId, ackAttemptId1, checkpointId));
+
+		assertTrue(pendingCheckpoint.isDiscarded());
+
+		// check that we've cleaned up the already acknowledged state
+		verify(triggerSubtaskState, times(1)).discardState();
+
+		StateHandle<?> ackSubtaskState = mock(StateHandle.class);
+		SerializedValue<StateHandle<?>> ackSerializedValue = mock(SerializedValue.class);
+		doReturn(ackSubtaskState).when(ackSerializedValue).deserializeValue(any(ClassLoader.class));
+
+		// late acknowledge message from the second ack vertex
+		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jobId, ackAttemptId2, checkpointId,
ackSerializedValue, 1L));
+
+		// check that we also cleaned up this state
+		verify(ackSubtaskState, times(1)).discardState();
+
+		// receive an acknowledge message from an unknown job
+		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(new JobID(), new ExecutionAttemptID(),
checkpointId, differentJobSerializedValue, 1L));
+
+		// we should not interfere with different jobs
+		verify(differentJobSubtaskState, never()).discardState();
+
+		StateHandle<?> unknownSubtaskState2 = mock(StateHandle.class);
+		SerializedValue<StateHandle<?>> unknownSerializedValue2 = mock(SerializedValue.class);
+		doReturn(unknownSubtaskState2).when(unknownSerializedValue2).deserializeValue(any(ClassLoader.class));
+
+		// receive an acknowledge message for an unknown vertex
+		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jobId, new ExecutionAttemptID(),
checkpointId, unknownSerializedValue2, 1L));
+
+		// we should discard acknowledge messages from an unknown vertex belonging to our job
+		verify(unknownSubtaskState2, times(1)).discardState();
+	}
+
 	@Test
 	public void testPeriodicTriggering() {
 		try {
@@ -1375,18 +1491,32 @@ public class CheckpointCoordinatorTest {
 	// ------------------------------------------------------------------------
 
 	private static ExecutionVertex mockExecutionVertex(ExecutionAttemptID attemptID) {
-		return mockExecutionVertex(attemptID, ExecutionState.RUNNING);
+		return mockExecutionVertex(attemptID, 1);
+	}
+
+	private static ExecutionVertex mockExecutionVertex(ExecutionAttemptID attemptId, int parallelism)
{
+		return mockExecutionVertex(attemptId, ExecutionState.RUNNING, parallelism);
 	}
 
 	private static ExecutionVertex mockExecutionVertex(ExecutionAttemptID attemptID, 
 														ExecutionState state, ExecutionState ... successiveStates) {
+		return mockExecutionVertex(attemptID, state, 1, successiveStates);
+	}
+
+	private static ExecutionVertex mockExecutionVertex(
+			ExecutionAttemptID attemptId,
+			ExecutionState state,
+			int parallelism,
+			ExecutionState ... successiveStates) {
+
 		final Execution exec = mock(Execution.class);
-		when(exec.getAttemptId()).thenReturn(attemptID);
+		when(exec.getAttemptId()).thenReturn(attemptId);
 		when(exec.getState()).thenReturn(state, successiveStates);
 
 		ExecutionVertex vertex = mock(ExecutionVertex.class);
 		when(vertex.getJobvertexId()).thenReturn(new JobVertexID());
 		when(vertex.getCurrentExecutionAttempt()).thenReturn(exec);
+		when(vertex.getTotalNumberOfParallelSubtasks()).thenReturn(parallelism);
 
 		return vertex;
 	}


Mime
View raw message