[FLINK-21451][coordination] Remove JobID from TaskExecutionState

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/TaskExecutionStateTransition.java

@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.executiongraph;
 
-import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
 import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.taskmanager.TaskExecutionState;
@@ -64,10 +63,6 @@ public ExecutionState getExecutionState() {
  return taskExecutionState.getExecutionState();
  }
 
- public JobID getJobID() {
- return taskExecutionState.getJobID();
- }
-
  public AccumulatorSnapshot getAccumulators() {
  return taskExecutionState.getAccumulators();
  }

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java

@@ -236,7 +236,6 @@ public void onMissingDeploymentsOf(
  for (ExecutionAttemptID executionAttemptId : executionAttemptIds) {
  schedulerNG.updateTaskExecutionState(
  new TaskExecutionState(
- jobGraph.getJobID(),
  executionAttemptId,
  ExecutionState.FAILED,
  new FlinkException(

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java

@@ -305,7 +305,7 @@ private ExecutionGraph createAndRestoreExecutionGraph(
  }
 
  newExecutionGraph.setInternalTaskFailuresListener(
- new UpdateSchedulerNgOnInternalFailuresListener(this, jobGraph.getJobID()));
+ new UpdateSchedulerNgOnInternalFailuresListener(this));
  newExecutionGraph.registerJobStatusListener(jobStatusListener);
  newExecutionGraph.start(mainThreadExecutor);
 

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/UpdateSchedulerNgOnInternalFailuresListener.java

@@ -19,7 +19,6 @@
 
 package org.apache.flink.runtime.scheduler;
 
-import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
 import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
@@ -35,13 +34,9 @@ public class UpdateSchedulerNgOnInternalFailuresListener implements InternalFail
 
  private final SchedulerNG schedulerNg;
 
- private final JobID jobId;
-
- public UpdateSchedulerNgOnInternalFailuresListener(
- final SchedulerNG schedulerNg, final JobID jobId) {
+ public UpdateSchedulerNgOnInternalFailuresListener(final SchedulerNG schedulerNg) {
 
  this.schedulerNg = checkNotNull(schedulerNg);
- this.jobId = checkNotNull(jobId);
  }
 
  @Override
@@ -52,7 +47,7 @@ public void notifyTaskFailure(
  final boolean releasePartitions) {
 
  final TaskExecutionState state =
- new TaskExecutionState(jobId, attemptId, ExecutionState.FAILED, t);
+ new TaskExecutionState(attemptId, ExecutionState.FAILED, t);
  schedulerNg.updateTaskExecutionState(
  new TaskExecutionStateTransition(state, cancelTask, releasePartitions));
  }

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java

@@ -609,7 +609,7 @@ public ExecutionGraph createExecutionGraphWithAvailableResources() throws Except
  executionGraph.transitionToRunning();
 
  executionGraph.setInternalTaskFailuresListener(
- new UpdateSchedulerNgOnInternalFailuresListener(this, jobInformation.getJobID()));
+ new UpdateSchedulerNgOnInternalFailuresListener(this));
 
  for (ExecutionVertex executionVertex : executionGraph.getAllExecutionVertices()) {
  final LogicalSlot assignedSlot =

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java

@@ -1749,7 +1749,6 @@ private void unregisterTaskAndNotifyFinalState(
  updateTaskExecutionState(
  jobMasterGateway,
  new TaskExecutionState(
- task.getJobID(),
  task.getExecutionId(),
  task.getExecutionState(),
  task.getFailureCause(),

flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java

@@ -745,7 +745,7 @@ private void doRun() {
 
  // notify everyone that we switched to running
  taskManagerActions.updateTaskExecutionState(
- new TaskExecutionState(jobId, executionId, ExecutionState.RUNNING));
+ new TaskExecutionState(executionId, ExecutionState.RUNNING));
 
  // make sure the user code classloader is accessible thread-locally
  executingThread.setContextClassLoader(userCodeClassLoader.asClassLoader());
@@ -1001,7 +1001,7 @@ private UserCodeClassLoader createUserCodeClassloader() throws Exception {
  private void notifyFinalState() {
  checkState(executionState.isTerminal());
  taskManagerActions.updateTaskExecutionState(
- new TaskExecutionState(jobId, executionId, executionState, failureCause));
+ new TaskExecutionState(executionId, executionState, failureCause));
  }
 
  private void notifyFatalError(String message, Throwable cause) {

flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskExecutionState.java

@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.taskmanager;
 
-import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
 import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
@@ -39,8 +38,6 @@ public class TaskExecutionState implements Serializable {
 
  private static final long serialVersionUID = 1L;
 
- private final JobID jobID;
-
  private final ExecutionAttemptID executionId;
 
  private final ExecutionState executionState;
@@ -55,53 +52,44 @@ public class TaskExecutionState implements Serializable {
  /**
  * Creates a new task execution state update, with no attached exception and no accumulators.
  *
- * @param jobID the ID of the job the task belongs to
  * @param executionId the ID of the task execution whose state is to be reported
  * @param executionState the execution state to be reported
  */
- public TaskExecutionState(
- JobID jobID, ExecutionAttemptID executionId, ExecutionState executionState) {
- this(jobID, executionId, executionState, null, null, null);
+ public TaskExecutionState(ExecutionAttemptID executionId, ExecutionState executionState) {
+ this(executionId, executionState, null, null, null);
  }
 
  /**
  * Creates a new task execution state update, with an attached exception but no accumulators.
  *
- * @param jobID the ID of the job the task belongs to
  * @param executionId the ID of the task execution whose state is to be reported
  * @param executionState the execution state to be reported
  */
  public TaskExecutionState(
- JobID jobID,
- ExecutionAttemptID executionId,
- ExecutionState executionState,
- Throwable error) {
- this(jobID, executionId, executionState, error, null, null);
+ ExecutionAttemptID executionId, ExecutionState executionState, Throwable error) {
+ this(executionId, executionState, error, null, null);
  }
 
  /**
  * Creates a new task execution state update, with an attached exception. This constructor may
  * never throw an exception.
  *
- * @param jobID the ID of the job the task belongs to
  * @param executionId the ID of the task execution whose state is to be reported
  * @param executionState the execution state to be reported
  * @param error an optional error
  * @param accumulators The flink and user-defined accumulators which may be null.
  */
  public TaskExecutionState(
- JobID jobID,
  ExecutionAttemptID executionId,
  ExecutionState executionState,
  Throwable error,
  AccumulatorSnapshot accumulators,
  IOMetrics ioMetrics) {
 
- if (jobID == null || executionId == null || executionState == null) {
+ if (executionId == null || executionState == null) {
  throw new NullPointerException();
  }
 
- this.jobID = jobID;
  this.executionId = executionId;
  this.executionState = executionState;
  if (error != null) {
@@ -148,15 +136,6 @@ public ExecutionState getExecutionState() {
  return this.executionState;
  }
 
- /**
- * The ID of the job the task belongs to
- *
- * @return the ID of the job the task belongs to
- */
- public JobID getJobID() {
- return this.jobID;
- }
-
  /** Gets flink and user-defined accumulators in serialized form. */
  public AccumulatorSnapshot getAccumulators() {
  return accumulators;
@@ -172,8 +151,7 @@ public IOMetrics getIOMetrics() {
  public boolean equals(Object obj) {
  if (obj instanceof TaskExecutionState) {
  TaskExecutionState other = (TaskExecutionState) obj;
- return other.jobID.equals(this.jobID)
- && other.executionId.equals(this.executionId)
+ return other.executionId.equals(this.executionId)
  && other.executionState == this.executionState
  && (other.throwable == null) == (this.throwable == null);
  } else {
@@ -183,16 +161,13 @@ public boolean equals(Object obj) {
 
  @Override
  public int hashCode() {
- return jobID.hashCode() + executionId.hashCode() + executionState.ordinal();
+ return executionId.hashCode() + executionState.ordinal();
  }
 
  @Override
  public String toString() {
  return String.format(
- "TaskExecutionState jobId=%s, executionId=%s, state=%s, error=%s",
- jobID,
- executionId,
- executionState,
- throwable == null ? "(null)" : throwable.toString());
+ "TaskExecutionState executionId=%s, state=%s, error=%s",
+ executionId, executionState, throwable == null ? "(null)" : throwable.toString());
  }
 }

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java

@@ -117,9 +117,7 @@ public void testShutdownCheckpointCoordinatorOnFinished() throws Exception {
  final Execution currentExecutionAttempt = executionVertex.getCurrentExecutionAttempt();
  scheduler.updateTaskExecutionState(
  new TaskExecutionState(
- graph.getJobID(),
- currentExecutionAttempt.getAttemptId(),
- ExecutionState.FINISHED));
+ currentExecutionAttempt.getAttemptId(), ExecutionState.FINISHED));
  }
 
  assertThat(graph.getTerminationFuture().get(), is(JobStatus.FINISHED));

flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraphTest.java

@@ -118,7 +118,6 @@ public static void setupExecutionGraph() throws Exception {
  scheduler.startScheduling();
  scheduler.updateTaskExecutionState(
  new TaskExecutionState(
- jobGraph.getJobID(),
  runtimeGraph
  .getAllExecutionVertices()
  .iterator()

flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java

@@ -340,7 +340,6 @@ public void testAccumulatorsAndMetricsForwarding() throws Exception {
 
  TaskExecutionState state =
  new TaskExecutionState(
- graph.getJobID(),
  execution1.getAttemptId(),
  ExecutionState.CANCELED,
  null,
@@ -364,7 +363,6 @@ public void testAccumulatorsAndMetricsForwarding() throws Exception {
 
  TaskExecutionState state2 =
  new TaskExecutionState(
- graph.getJobID(),
  execution2.getAttemptId(),
  ExecutionState.FAILED,
  null,
@@ -491,9 +489,9 @@ public void testNoResourceAvailableFailure() throws Exception {
  .getCurrentExecutionAttempt()
  .getAttemptId();
  scheduler.updateTaskExecutionState(
- new TaskExecutionState(jobId, attemptID, ExecutionState.RUNNING));
+ new TaskExecutionState(attemptID, ExecutionState.RUNNING));
  scheduler.updateTaskExecutionState(
- new TaskExecutionState(jobId, attemptID, ExecutionState.FINISHED, null));
+ new TaskExecutionState(attemptID, ExecutionState.FINISHED, null));
 
  assertEquals(JobStatus.FAILED, eg.getState());
  }

flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java

@@ -89,9 +89,7 @@ public void testStrategyNotifiedOfFinishedVerticesAndResultsRespected() throws E
  getCurrentExecution(sourceVertex, executionGraph);
  scheduler.updateTaskExecutionState(
  new TaskExecutionState(
- executionGraph.getJobID(),
- sourceExecution.getAttemptId(),
- ExecutionState.FINISHED));
+ sourceExecution.getAttemptId(), ExecutionState.FINISHED));
  assertThat(releasedPartitions, empty());
  });
 
@@ -103,9 +101,7 @@ public void testStrategyNotifiedOfFinishedVerticesAndResultsRespected() throws E
  getCurrentExecution(operatorVertex, executionGraph);
  scheduler.updateTaskExecutionState(
  new TaskExecutionState(
- executionGraph.getJobID(),
- operatorExecution.getAttemptId(),
- ExecutionState.FINISHED));
+ operatorExecution.getAttemptId(), ExecutionState.FINISHED));
  assertThat(releasedPartitions, hasSize(1));
  assertThat(
  releasedPartitions.remove(),
@@ -127,9 +123,7 @@ public void testStrategyNotifiedOfFinishedVerticesAndResultsRespected() throws E
  final Execution sinkExecution = getCurrentExecution(sinkVertex, executionGraph);
  scheduler.updateTaskExecutionState(
  new TaskExecutionState(
- executionGraph.getJobID(),
- sinkExecution.getAttemptId(),
- ExecutionState.FINISHED));
+ sinkExecution.getAttemptId(), ExecutionState.FINISHED));
 
  assertThat(releasedPartitions, hasSize(1));
  assertThat(
@@ -187,9 +181,7 @@ public void testStrategyNotifiedOfUnFinishedVertices() throws Exception {
  // consumer o1 was not finished
  scheduler.updateTaskExecutionState(
  new TaskExecutionState(
- executionGraph.getJobID(),
- sourceExecution.getAttemptId(),
- ExecutionState.FINISHED));
+ sourceExecution.getAttemptId(), ExecutionState.FINISHED));
  assertThat(releasedPartitions, empty());
  });
 
@@ -207,9 +199,7 @@ public void testStrategyNotifiedOfUnFinishedVertices() throws Exception {
  }
  scheduler.updateTaskExecutionState(
  new TaskExecutionState(
- executionGraph.getJobID(),
- operator1Execution.getAttemptId(),
- ExecutionState.FINISHED));
+ operator1Execution.getAttemptId(), ExecutionState.FINISHED));
  assertThat(releasedPartitions, empty());
  });
 
@@ -221,9 +211,7 @@ public void testStrategyNotifiedOfUnFinishedVertices() throws Exception {
  // finished
  scheduler.updateTaskExecutionState(
  new TaskExecutionState(
- executionGraph.getJobID(),
- operator2Execution.getAttemptId(),
- ExecutionState.FINISHED));
+ operator2Execution.getAttemptId(), ExecutionState.FINISHED));
  assertThat(releasedPartitions, empty());
  });
 
@@ -243,9 +231,7 @@ public void testStrategyNotifiedOfUnFinishedVertices() throws Exception {
  // finish o3; this should not result in any release calls since o2 was reset
  scheduler.updateTaskExecutionState(
  new TaskExecutionState(
- executionGraph.getJobID(),
- operator3Execution.getAttemptId(),
- ExecutionState.FINISHED));
+ operator3Execution.getAttemptId(), ExecutionState.FINISHED));
  assertThat(releasedPartitions, empty());
  });
  }

flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/UpdatePartitionConsumersTest.java

@@ -189,9 +189,6 @@ public void testUpdatePartitionConsumers() throws Exception {
  private void updateState(
  SchedulerBase scheduler, ExecutionVertex vertex, ExecutionState state) {
  scheduler.updateTaskExecutionState(
- new TaskExecutionState(
- jobGraph.getJobID(),
- vertex.getCurrentExecutionAttempt().getAttemptId(),
- state));
+ new TaskExecutionState(vertex.getCurrentExecutionAttempt().getAttemptId(), state));
  }
 }

flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterPartitionReleaseTest.java

@@ -164,9 +164,7 @@ private void testPartitionReleaseOrPromotionOnJobTermination(
  taskDeploymentDescriptorFuture.get();
  jobMasterGateway.updateTaskExecutionState(
  new TaskExecutionState(
- taskDeploymentDescriptor.getJobId(),
- taskDeploymentDescriptor.getExecutionAttemptId(),
- finalExecutionState));
+ taskDeploymentDescriptor.getExecutionAttemptId(), finalExecutionState));
 
  assertThat(
  taskExecutorCallSelector.apply(testSetup).get(),