You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by tillrohrmann <gi...@git.apache.org> on 2016/04/14 11:18:29 UTC
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
GitHub user tillrohrmann opened a pull request:
https://github.com/apache/flink/pull/1883
[FLINK-3756] [state] Add state hierarchy to CheckpointCoordinator
This commit introduces a state hierarchy for the StateForTask objects kept
at the CheckpointCoordinator. Task states are now grouped together if they
belong to the same ExecutionJobVertex. The StateForTask objects are now
stored in so called StateForTaskGroup objects. The StateForTaskGroup object
can also store the key group state handles associated to a ExecutionJobVertex.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tillrohrmann/flink stateHierarchy
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/1883.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1883
----
commit c9217e7950397e94faadaec31142538311f8c050
Author: Till Rohrmann <tr...@apache.org>
Date: 2016-04-08T13:20:06Z
[FLINK-3756] [state] Add state hierarchy to CheckpointCoordinator
This commit introduces a state hierarchy for the StateForTask objects kept
at the CheckpointCoordinator. Task states are now grouped together if they
belong to the same ExecutionJobVertex. The StateForTask objects are now
stored in so called StateForTaskGroup objects. The StateForTaskGroup object
can also store the key group state handles associated to a ExecutionJobVertex.
Adapt restore methods of CheckpointCoordinator and SavepointCoordinator
Add state size computation
Add comments
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the pull request:
https://github.com/apache/flink/pull/1883#issuecomment-214683701
I agree. Our terminology is a mess an should be cleaned up. Will do this as part of my other refactorings.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r60906533
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java ---
@@ -781,43 +803,69 @@ public boolean restoreLatestCheckpointedState(
long recoveryTimestamp = System.currentTimeMillis();
- if (allOrNothingState) {
- Map<ExecutionJobVertex, Integer> stateCounts = new HashMap<ExecutionJobVertex, Integer>();
+ for (Map.Entry<JobVertexID, StateForTaskGroup> taskGroupStateEntry: latest.getTaskGroupStates().entrySet()) {
+ StateForTaskGroup taskGroupState = taskGroupStateEntry.getValue();
+ ExecutionJobVertex executionJobVertex = tasks.get(taskGroupStateEntry.getKey());
+
+ if (executionJobVertex != null) {
+ // check that we only restore the state if the parallelism has not been changed
+ if (taskGroupState.getParallelism() != executionJobVertex.getParallelism()) {
+ throw new RuntimeException("Cannot restore the latest checkpoint because " +
+ "the parallelism changed. The operator" + executionJobVertex.getJobVertexId() +
+ " has parallelism " + executionJobVertex.getParallelism() + " whereas the corresponding" +
+ "state object has a parallelism of " + taskGroupState.getParallelism());
+ }
- for (StateForTask state : latest.getStates()) {
- ExecutionJobVertex vertex = tasks.get(state.getOperatorId());
- Execution exec = vertex.getTaskVertices()[state.getSubtask()].getCurrentExecutionAttempt();
- exec.setInitialState(state.getState(), recoveryTimestamp);
+ int counter = 0;
- Integer count = stateCounts.get(vertex);
- if (count != null) {
- stateCounts.put(vertex, count+1);
- } else {
- stateCounts.put(vertex, 1);
+ List<Set<Integer>> keyGroupPartitions = createKeyGroupPartitions(numberKeyGroups, executionJobVertex.getParallelism());
+
+ for (int i = 0; i < executionJobVertex.getParallelism(); i++) {
+ StateForTask stateForTask = taskGroupState.getState(i);
+ SerializedValue<StateHandle<?>> state = null;
+
+ if (stateForTask != null) {
+ // count the number of executions for which we set a state
+ counter++;
+ state = stateForTask.getState();
+ }
+
+ Map<Integer, SerializedValue<StateHandle<?>>> kvStateForTaskMap = taskGroupState.getUnwrappedKvStates(keyGroupPartitions.get(i));
+
+ Execution currentExecutionAttempt = executionJobVertex.getTaskVertices()[i].getCurrentExecutionAttempt();
+ currentExecutionAttempt.setInitialState(state, kvStateForTaskMap, recoveryTimestamp);
}
- }
- // validate that either all task vertices have state, or none
- for (Map.Entry<ExecutionJobVertex, Integer> entry : stateCounts.entrySet()) {
- ExecutionJobVertex vertex = entry.getKey();
- if (entry.getValue() != vertex.getParallelism()) {
- throw new IllegalStateException(
- "The checkpoint contained state only for a subset of tasks for vertex " + vertex);
+ if (allOrNothingState && counter > 0 && counter < executionJobVertex.getParallelism()) {
+ throw new IllegalStateException("The checkpoint contained state only for " +
+ "a subset of tasks for vertex " + executionJobVertex);
}
- }
- }
- else {
- for (StateForTask state : latest.getStates()) {
- ExecutionJobVertex vertex = tasks.get(state.getOperatorId());
- Execution exec = vertex.getTaskVertices()[state.getSubtask()].getCurrentExecutionAttempt();
- exec.setInitialState(state.getState(), recoveryTimestamp);
+ } else {
+ throw new IllegalStateException("There is no execution job vertex for the job" +
+ " vertex ID " + taskGroupStateEntry.getKey());
}
}
return true;
}
}
+ protected List<Set<Integer>> createKeyGroupPartitions(int numberKeyGroups, int parallelism) {
--- End diff --
That is true. Will add it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1883#issuecomment-214678348
Unrelated, but shouldn't we consolidate the code in `SavepointCoordinator.restoreSavepoint()` and `CheckpointCoordiantor.restoreLatestCheckpointedState`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r60903230
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java ---
@@ -781,43 +803,69 @@ public boolean restoreLatestCheckpointedState(
long recoveryTimestamp = System.currentTimeMillis();
- if (allOrNothingState) {
- Map<ExecutionJobVertex, Integer> stateCounts = new HashMap<ExecutionJobVertex, Integer>();
+ for (Map.Entry<JobVertexID, StateForTaskGroup> taskGroupStateEntry: latest.getTaskGroupStates().entrySet()) {
+ StateForTaskGroup taskGroupState = taskGroupStateEntry.getValue();
+ ExecutionJobVertex executionJobVertex = tasks.get(taskGroupStateEntry.getKey());
+
+ if (executionJobVertex != null) {
+ // check that we only restore the state if the parallelism has not been changed
+ if (taskGroupState.getParallelism() != executionJobVertex.getParallelism()) {
+ throw new RuntimeException("Cannot restore the latest checkpoint because " +
+ "the parallelism changed. The operator" + executionJobVertex.getJobVertexId() +
+ " has parallelism " + executionJobVertex.getParallelism() + " whereas the corresponding" +
+ "state object has a parallelism of " + taskGroupState.getParallelism());
+ }
- for (StateForTask state : latest.getStates()) {
- ExecutionJobVertex vertex = tasks.get(state.getOperatorId());
- Execution exec = vertex.getTaskVertices()[state.getSubtask()].getCurrentExecutionAttempt();
- exec.setInitialState(state.getState(), recoveryTimestamp);
+ int counter = 0;
- Integer count = stateCounts.get(vertex);
- if (count != null) {
- stateCounts.put(vertex, count+1);
- } else {
- stateCounts.put(vertex, 1);
+ List<Set<Integer>> keyGroupPartitions = createKeyGroupPartitions(numberKeyGroups, executionJobVertex.getParallelism());
+
+ for (int i = 0; i < executionJobVertex.getParallelism(); i++) {
+ StateForTask stateForTask = taskGroupState.getState(i);
+ SerializedValue<StateHandle<?>> state = null;
+
+ if (stateForTask != null) {
+ // count the number of executions for which we set a state
+ counter++;
+ state = stateForTask.getState();
+ }
+
+ Map<Integer, SerializedValue<StateHandle<?>>> kvStateForTaskMap = taskGroupState.getUnwrappedKvStates(keyGroupPartitions.get(i));
+
+ Execution currentExecutionAttempt = executionJobVertex.getTaskVertices()[i].getCurrentExecutionAttempt();
+ currentExecutionAttempt.setInitialState(state, kvStateForTaskMap, recoveryTimestamp);
}
- }
- // validate that either all task vertices have state, or none
- for (Map.Entry<ExecutionJobVertex, Integer> entry : stateCounts.entrySet()) {
- ExecutionJobVertex vertex = entry.getKey();
- if (entry.getValue() != vertex.getParallelism()) {
- throw new IllegalStateException(
- "The checkpoint contained state only for a subset of tasks for vertex " + vertex);
+ if (allOrNothingState && counter > 0 && counter < executionJobVertex.getParallelism()) {
+ throw new IllegalStateException("The checkpoint contained state only for " +
+ "a subset of tasks for vertex " + executionJobVertex);
}
- }
- }
- else {
- for (StateForTask state : latest.getStates()) {
- ExecutionJobVertex vertex = tasks.get(state.getOperatorId());
- Execution exec = vertex.getTaskVertices()[state.getSubtask()].getCurrentExecutionAttempt();
- exec.setInitialState(state.getState(), recoveryTimestamp);
+ } else {
+ throw new IllegalStateException("There is no execution job vertex for the job" +
+ " vertex ID " + taskGroupStateEntry.getKey());
}
}
return true;
}
}
+ protected List<Set<Integer>> createKeyGroupPartitions(int numberKeyGroups, int parallelism) {
--- End diff --
I think it would be great to add a high-level comment to this method
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r59712024
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java ---
@@ -139,13 +144,43 @@ public boolean acknowledgeTask(
ExecutionVertex vertex = notYetAcknowledgedTasks.remove(attemptID);
if (vertex != null) {
- if (state != null) {
- collectedStates.add(new StateForTask(
- state,
- stateSize,
- vertex.getJobvertexId(),
+ if (state != null || kvState != null) {
+
+ JobVertexID jobVertexID = vertex.getJobvertexId();
+
+ StateForTaskGroup taskGroupState;
+
+ if (taskGroupStates.containsKey(jobVertexID)) {
+ taskGroupState = taskGroupStates.get(jobVertexID);
+ } else {
+ taskGroupState = new StateForTaskGroup(jobVertexID, vertex.getTotalNumberOfParallelSubtasks());
+ taskGroupStates.put(jobVertexID, taskGroupState);
+ }
+
+ long timestamp = System.currentTimeMillis() - checkpointTimestamp;
+
+ if (state != null) {
+ taskGroupState.putState(
vertex.getParallelSubtaskIndex(),
- System.currentTimeMillis() - checkpointTimestamp));
+ new StateForTask(
+ state,
+ stateSize,
+ timestamp
+ )
+ );
+ }
+
+ if (kvState != null) {
+ for (Map.Entry<Integer, SerializedValue<StateHandle<?>>> entry : kvState.entrySet()) {
+ taskGroupState.putKvState(
+ entry.getKey(),
+ new KvStateForTasks(
+ entry.getValue(),
+ 0L,
--- End diff --
Because at the moment the key-value state is still contained in the `state` object. This will be changed with the next PR introducing key groups for key-value state.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the pull request:
https://github.com/apache/flink/pull/1883#issuecomment-214293767
Very nice, Till! I like where this is going and am really looking forward to the follow up changes.
As this is work in progress, the most important changes are in the restore methods of the checkpoint/savepoint coordinator and those look good to me.
The Travis test failures are unrelated, so +1 to merge this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the pull request:
https://github.com/apache/flink/pull/1883#issuecomment-214686210
Yes, it was introduced to give detailed error message, which were helpful in the past. We can certainly consolidate this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/1883
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the pull request:
https://github.com/apache/flink/pull/1883#issuecomment-214311164
Thanks for the review @uce. I've addressed your comments. Will merge this PR then.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r60906496
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java ---
@@ -640,15 +650,27 @@ public boolean receiveAcknowledgeMessage(AcknowledgeCheckpoint message) throws E
if (checkpoint != null && !checkpoint.isDiscarded()) {
isPendingCheckpoint = true;
- if (checkpoint.acknowledgeTask(message.getTaskExecutionId(), message.getState(), message.getStateSize())) {
+ if (checkpoint.acknowledgeTask(
+ message.getTaskExecutionId(),
+ message.getState(),
+ message.getStateSize(),
+ null)) { // TODO: Give KV-state to the acknowledgeTask method
--- End diff --
Yes, exactly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r60903617
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java ---
@@ -130,7 +134,8 @@ public CompletedCheckpoint toCompletedCheckpoint() {
public boolean acknowledgeTask(
ExecutionAttemptID attemptID,
SerializedValue<StateHandle<?>> state,
- long stateSize) {
+ long stateSize,
+ Map<Integer, SerializedValue<StateHandle<?>>> kvState) {
--- End diff --
My personal preference, but let's add `kvState` after the `state` argument?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r60907150
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java ---
@@ -74,20 +77,56 @@ public long getDuration() {
return duration;
}
- public List<StateForTask> getStates() {
- return states;
+ public long getStateSize() {
+ long result = 0L;
+
+ for (StateForTaskGroup stateForTaskGroup: taskGroupStates.values()) {
+ result += stateForTaskGroup.getStateSize();
+ }
+
+ return result;
+ }
+
+ public Map<JobVertexID, StateForTaskGroup> getTaskGroupStates() {
+ return taskGroupStates;
+ }
+
+ public StateForTaskGroup getTaskGroupState(JobVertexID jobVertexID) {
+ return taskGroupStates.get(jobVertexID);
}
// --------------------------------------------------------------------------------------------
public void discard(ClassLoader userClassLoader) {
- for(StateForTask state: states){
+ for (StateForTaskGroup state: taskGroupStates.values()) {
state.discard(userClassLoader);
}
- states.clear();
+
+ taskGroupStates.clear();
}
// --------------------------------------------------------------------------------------------
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof CompletedCheckpoint) {
+ CompletedCheckpoint other = (CompletedCheckpoint) obj;
+
+ return job.equals(other.job) && checkpointID == other.checkpointID &&
+ timestamp == other.timestamp && duration == other.duration &&
+ taskGroupStates.equals(other.taskGroupStates);
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public int hashCode() {
--- End diff --
I added it because a test case was comparing `CompletedCheckpoints` by comparing the `taskGroupStates` individually. I thought this would be better to do in the `equals` method.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r59687315
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java ---
@@ -139,13 +144,43 @@ public boolean acknowledgeTask(
ExecutionVertex vertex = notYetAcknowledgedTasks.remove(attemptID);
if (vertex != null) {
- if (state != null) {
- collectedStates.add(new StateForTask(
- state,
- stateSize,
- vertex.getJobvertexId(),
+ if (state != null || kvState != null) {
+
+ JobVertexID jobVertexID = vertex.getJobvertexId();
+
+ StateForTaskGroup taskGroupState;
+
+ if (taskGroupStates.containsKey(jobVertexID)) {
+ taskGroupState = taskGroupStates.get(jobVertexID);
+ } else {
+ taskGroupState = new StateForTaskGroup(jobVertexID, vertex.getTotalNumberOfParallelSubtasks());
+ taskGroupStates.put(jobVertexID, taskGroupState);
+ }
+
+ long timestamp = System.currentTimeMillis() - checkpointTimestamp;
+
+ if (state != null) {
+ taskGroupState.putState(
vertex.getParallelSubtaskIndex(),
- System.currentTimeMillis() - checkpointTimestamp));
+ new StateForTask(
+ state,
+ stateSize,
+ timestamp
+ )
+ );
+ }
+
+ if (kvState != null) {
+ for (Map.Entry<Integer, SerializedValue<StateHandle<?>>> entry : kvState.entrySet()) {
+ taskGroupState.putKvState(
+ entry.getKey(),
+ new KvStateForTasks(
+ entry.getValue(),
+ 0L,
--- End diff --
Why is this `0L`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r60906793
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java ---
@@ -130,7 +134,8 @@ public CompletedCheckpoint toCompletedCheckpoint() {
public boolean acknowledgeTask(
ExecutionAttemptID attemptID,
SerializedValue<StateHandle<?>> state,
- long stateSize) {
+ long stateSize,
+ Map<Integer, SerializedValue<StateHandle<?>>> kvState) {
--- End diff --
The thing is that the `kvState` also needs the state size information. Thus, the method should get another parameter `Map<Integer, Long> kvStateSizes` and then I think the state size should be grouped together with the serialized state handle.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r60903497
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java ---
@@ -74,20 +77,56 @@ public long getDuration() {
return duration;
}
- public List<StateForTask> getStates() {
- return states;
+ public long getStateSize() {
+ long result = 0L;
+
+ for (StateForTaskGroup stateForTaskGroup: taskGroupStates.values()) {
+ result += stateForTaskGroup.getStateSize();
+ }
+
+ return result;
+ }
+
+ public Map<JobVertexID, StateForTaskGroup> getTaskGroupStates() {
+ return taskGroupStates;
+ }
+
+ public StateForTaskGroup getTaskGroupState(JobVertexID jobVertexID) {
+ return taskGroupStates.get(jobVertexID);
}
// --------------------------------------------------------------------------------------------
public void discard(ClassLoader userClassLoader) {
- for(StateForTask state: states){
+ for (StateForTaskGroup state: taskGroupStates.values()) {
state.discard(userClassLoader);
}
- states.clear();
+
+ taskGroupStates.clear();
}
// --------------------------------------------------------------------------------------------
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof CompletedCheckpoint) {
+ CompletedCheckpoint other = (CompletedCheckpoint) obj;
+
+ return job.equals(other.job) && checkpointID == other.checkpointID &&
+ timestamp == other.timestamp && duration == other.duration &&
+ taskGroupStates.equals(other.taskGroupStates);
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public int hashCode() {
--- End diff --
Out of curiosity, did you add this, because the default implementation was a problem somewhere (also for `PendingCheckpoint`)? Both looks good though.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1883#issuecomment-214683587
👍 good to merge, then
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the pull request:
https://github.com/apache/flink/pull/1883#issuecomment-214672764
Failing test is unrelated. Merging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1883#issuecomment-214683500
Also, I think we should rename the task business classes to accurately reflect what they are for. But this can be done once the rest of the state reworking is in.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r60906912
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateForTask.java ---
@@ -131,11 +108,13 @@ else if (o instanceof StateForTask) {
@Override
public int hashCode() {
- return state.hashCode() + 31 * operatorId.hashCode() + 43 * subtask;
+ return (int) (this.stateSize ^ this.stateSize >>> 32) +
+ 31 * ((int) (this.duration ^ this.duration >>> 32) +
+ 31 * state.hashCode());
}
@Override
public String toString() {
- return String.format("StateForTask %s-%d : %s", operatorId, subtask, state);
+ return String.format("StateForTask(%s)", state);
--- End diff --
True this is not so helpful. Will take a look what else one could return here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/flink/pull/1883#issuecomment-214680409
Also, we'll still call `StateForTask` like that, even though it's more like an operator state? And `KvStateForTasks` has the s at the end because it is theoretically for several tasks/subtasks/operators? Which wouldn't really work, right? 😅
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r60903817
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateForTask.java ---
@@ -131,11 +108,13 @@ else if (o instanceof StateForTask) {
@Override
public int hashCode() {
- return state.hashCode() + 31 * operatorId.hashCode() + 43 * subtask;
+ return (int) (this.stateSize ^ this.stateSize >>> 32) +
+ 31 * ((int) (this.duration ^ this.duration >>> 32) +
+ 31 * state.hashCode());
}
@Override
public String toString() {
- return String.format("StateForTask %s-%d : %s", operatorId, subtask, state);
+ return String.format("StateForTask(%s)", state);
--- End diff --
This was there before as well, but `toString` will always return `StateForTask(SerializedValue)`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the pull request:
https://github.com/apache/flink/pull/1883#issuecomment-214680362
I agree that this is duplicated code. As far as I can tell, this was introduced in order to output different error messages in the savepoint restore and the checkpoint restore case and to support the `allOrNothingState` checkpoint recovery. The latter seems to be only introduced for testing purposes.
Independent on our decision, I think this should be addressed as a separate issue.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3756] [state] Add state hierarchy to Ch...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1883#discussion_r60903126
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java ---
@@ -640,15 +650,27 @@ public boolean receiveAcknowledgeMessage(AcknowledgeCheckpoint message) throws E
if (checkpoint != null && !checkpoint.isDiscarded()) {
isPendingCheckpoint = true;
- if (checkpoint.acknowledgeTask(message.getTaskExecutionId(), message.getState(), message.getStateSize())) {
+ if (checkpoint.acknowledgeTask(
+ message.getTaskExecutionId(),
+ message.getState(),
+ message.getStateSize(),
+ null)) { // TODO: Give KV-state to the acknowledgeTask method
--- End diff --
This will be resolved as part of your follow-up changes, right? :-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---