[FLINK-11649] Remove legacy JobInfo class

flink-libraries/flink-ml/src/test/resources/logback-test.xml

@@ -34,9 +34,8 @@
     <logger name="org.apache.flink.runtime.operators.BatchTask" level="OFF"/>
     <logger name="org.apache.flink.runtime.client.JobClient" level="OFF"/>
     <logger name="org.apache.flink.runtime.taskmanager.Task" level="OFF"/>
-    <logger name="org.apache.flink.runtime.jobmanager.JobManager" level="OFF"/>
     <logger name="org.apache.flink.runtime.testingUtils" level="OFF"/>
     <logger name="org.apache.flink.runtime.executiongraph.ExecutionGraph" level="OFF"/>
     <logger name="org.apache.flink.runtime.jobmanager.EventCollector" level="OFF"/>
     <logger name="org.apache.flink.runtime.instance.InstanceManager" level="OFF"/>
-</configuration>
+</configuration>

flink-libraries/flink-python/src/test/resources/logback-test.xml

@@ -34,9 +34,8 @@
     <logger name="org.apache.flink.runtime.operators.BatchTask" level="OFF"/>
     <logger name="org.apache.flink.runtime.client.JobClient" level="OFF"/>
     <logger name="org.apache.flink.runtime.taskmanager.Task" level="OFF"/>
-    <logger name="org.apache.flink.runtime.jobmanager.JobManager" level="OFF"/>
     <logger name="org.apache.flink.runtime.testingUtils" level="OFF"/>
     <logger name="org.apache.flink.runtime.executiongraph.ExecutionGraph" level="OFF"/>
     <logger name="org.apache.flink.runtime.jobmanager.EventCollector" level="OFF"/>
     <logger name="org.apache.flink.runtime.instance.InstanceManager" level="OFF"/>
-</configuration>
+</configuration>

flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/SubmittedJobGraph.java

@@ -21,14 +21,12 @@
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 
-import javax.annotation.Nullable;
-
 import java.io.Serializable;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
- * A recoverable {@link JobGraph} and {@link JobInfo}.
+ * A recoverable {@link JobGraph}.
  */
 public class SubmittedJobGraph implements Serializable {
 
@@ -37,30 +35,13 @@ public class SubmittedJobGraph implements Serializable {
 	/** The submitted {@link JobGraph}. */
 	private final JobGraph jobGraph;
 
-	/** The {@link JobInfo}. */
-	private final JobInfo jobInfo;
-
 	/**
 	 * Creates a {@link SubmittedJobGraph}.
 	 *
 	 * @param jobGraph The submitted {@link JobGraph}
 	 */
 	public SubmittedJobGraph(JobGraph jobGraph) {
-		this(jobGraph, null);
-	}
-
-	/**
-	 * Creates a {@link SubmittedJobGraph}.
-	 *
-	 * @param jobGraph The submitted {@link JobGraph}
-	 * @param jobInfo  The {@link JobInfo}
-	 *
-	 * @deprecated FLIP-6 code should use {@link #SubmittedJobGraph(JobGraph)}
-	 */
-	@Deprecated
-	public SubmittedJobGraph(JobGraph jobGraph, @Nullable JobInfo jobInfo) {
 		this.jobGraph = checkNotNull(jobGraph, "Job graph");
-		this.jobInfo = jobInfo;
 	}
 
 	/**
@@ -77,18 +58,8 @@ public JobID getJobId() {
 		return jobGraph.getJobID();
 	}
 
-	/**
-	 * Returns the {@link JobInfo} of the client who submitted the {@link JobGraph}.
-	 *
-	 * @deprecated FLIP-6 code should not use this method because it will always return null.
-	 */
-	@Deprecated
-	public JobInfo getJobInfo() throws Exception {
-		return jobInfo;
-	}
-
 	@Override
 	public String toString() {
-		return String.format("SubmittedJobGraph(%s, %s)", jobGraph.getJobID(), jobInfo);
+		return String.format("SubmittedJobGraph(%s)", jobGraph.getJobID());
 	}
 }

flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/AcknowledgeCheckpoint.java

@@ -25,14 +25,14 @@
 import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
 
 /**
- * This message is sent from the {@link org.apache.flink.runtime.taskmanager.TaskManager} to the
- * {@link org.apache.flink.runtime.jobmanager.JobManager} to signal that the checkpoint of an
+ * This message is sent from the {@link org.apache.flink.runtime.taskexecutor.TaskExecutor} to the
+ * {@link org.apache.flink.runtime.jobmaster.JobMaster} to signal that the checkpoint of an
  * individual task is completed.
- * <p>
+ *
  * <p>This message may carry the handle to the task's chained operator state and the key group
  * state.
  */
-public class AcknowledgeCheckpoint extends AbstractCheckpointMessage implements java.io.Serializable {
+public class AcknowledgeCheckpoint extends AbstractCheckpointMessage {
 
 	private static final long serialVersionUID = -7606214777192401493L;
 

flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/DeclineCheckpoint.java

@@ -29,16 +29,16 @@
 import org.apache.flink.util.SerializedThrowable;
 
 /**
- * This message is sent from the {@link org.apache.flink.runtime.taskmanager.TaskManager} to the
- * {@link org.apache.flink.runtime.jobmanager.JobManager} to tell the checkpoint coordinator
+ * This message is sent from the {@link org.apache.flink.runtime.taskexecutor.TaskExecutor} to the
+ * {@link org.apache.flink.runtime.jobmaster.JobMaster} to tell the checkpoint coordinator
  * that a checkpoint request could not be heeded. This can happen if a Task is already in
  * RUNNING state but is internally not yet ready to perform checkpoints.
  */
 public class DeclineCheckpoint extends AbstractCheckpointMessage implements java.io.Serializable {
 
 	private static final long serialVersionUID = 2094094662279578953L;
 
-	/** The reason why the checkpoint was declined */
+	/** The reason why the checkpoint was declined. */
 	private final Throwable reason;
 
 	public DeclineCheckpoint(JobID job, ExecutionAttemptID taskExecutionId, long checkpointId) {
@@ -47,15 +47,14 @@ public DeclineCheckpoint(JobID job, ExecutionAttemptID taskExecutionId, long che
 
 	public DeclineCheckpoint(JobID job, ExecutionAttemptID taskExecutionId, long checkpointId, Throwable reason) {
 		super(job, taskExecutionId, checkpointId);
-		
+
 		if (reason == null ||
 			reason.getClass() == AlignmentLimitExceededException.class ||
 			reason.getClass() == CheckpointDeclineOnCancellationBarrierException.class ||
 			reason.getClass() == CheckpointDeclineSubsumedException.class ||
 			reason.getClass() == CheckpointDeclineTaskNotCheckpointingException.class ||
 			reason.getClass() == CheckpointDeclineTaskNotReadyException.class ||
-			reason.getClass() == InputEndOfStreamException.class)
-		{
+			reason.getClass() == InputEndOfStreamException.class) {
 			// null or known common exceptions that cannot reference any dynamically loaded code
 			this.reason = reason;
 		} else {
@@ -68,7 +67,7 @@ public DeclineCheckpoint(JobID job, ExecutionAttemptID taskExecutionId, long che
 
 	/**
 	 * Gets the reason why the checkpoint was declined.
-	 * 
+	 *
 	 * @return The reason why the checkpoint was declined
 	 */
 	public Throwable getReason() {

flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/NotifyCheckpointComplete.java

@@ -22,15 +22,15 @@
 import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
 
 /**
- * This message is sent from the {@link org.apache.flink.runtime.jobmanager.JobManager} to the
- * {@link org.apache.flink.runtime.taskmanager.TaskManager} to tell a task that the checkpoint
+ * This message is sent from the {@link org.apache.flink.runtime.jobmaster.JobMaster} to the
+ * {@link org.apache.flink.runtime.taskexecutor.TaskExecutor} to tell a task that the checkpoint
  * has been confirmed and that the task can commit the checkpoint to the outside world.
  */
 public class NotifyCheckpointComplete extends AbstractCheckpointMessage implements java.io.Serializable {
 
 	private static final long serialVersionUID = 2094094662279578953L;
 
-	/** The timestamp associated with the checkpoint */
+	/** The timestamp associated with the checkpoint. */
 	private final long timestamp;
 
 	public NotifyCheckpointComplete(JobID job, ExecutionAttemptID taskExecutionId, long checkpointId, long timestamp) {
@@ -67,7 +67,10 @@ else if (o instanceof NotifyCheckpointComplete) {
 
 	@Override
 	public String toString() {
-		return String.format("ConfirmCheckpoint %d for (%s/%s)", 
-				getCheckpointId(), getJob(), getTaskExecutionId());
+		return String.format(
+			"ConfirmCheckpoint %d for (%s/%s)",
+			getCheckpointId(),
+			getJob(),
+			getTaskExecutionId());
 	}
 }

flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/TriggerCheckpoint.java

@@ -18,22 +18,22 @@
 
 package org.apache.flink.runtime.messages.checkpoint;
 
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
 import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
 
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
 /**
- * This message is sent from the {@link org.apache.flink.runtime.jobmanager.JobManager} to the
- * {@link org.apache.flink.runtime.taskmanager.TaskManager} to tell a certain task to trigger its
+ * This message is sent from the {@link org.apache.flink.runtime.jobmaster.JobMaster} to the
+ * {@link org.apache.flink.runtime.taskexecutor.TaskExecutor} to tell a certain task to trigger its
  * checkpoint.
  */
-public class TriggerCheckpoint extends AbstractCheckpointMessage implements java.io.Serializable {
+public class TriggerCheckpoint extends AbstractCheckpointMessage {
 
 	private static final long serialVersionUID = 2094094662279578953L;
-	
-	/** The timestamp associated with the checkpoint */
+
+	/** The timestamp associated with the checkpoint. */
 	private final long timestamp;
 
 	/** Options for how to perform the checkpoint. */
@@ -52,7 +52,7 @@ public TriggerCheckpoint(
 	}
 
 	// --------------------------------------------------------------------------------------------
-	
+
 	public long getTimestamp() {
 		return timestamp;
 	}
@@ -84,7 +84,11 @@ else if (o instanceof TriggerCheckpoint) {
 
 	@Override
 	public String toString() {
-		return String.format("Trigger Checkpoint %d@%d for (%s/%s)", 
-				getCheckpointId(), getTimestamp(), getJob(), getTaskExecutionId());
+		return String.format(
+			"Trigger Checkpoint %d@%d for (%s/%s)",
+			getCheckpointId(),
+			getTimestamp(),
+			getJob(),
+			getTaskExecutionId());
 	}
 }

flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/package-info.java

@@ -17,8 +17,8 @@
  */
 
 /**
- * This package contains the messages that are sent between {@link org.apache.flink.runtime.jobmanager.JobManager}
- * and {@link org.apache.flink.runtime.taskmanager.TaskManager} to coordinate the checkpoint snapshots of the
+ * This package contains the messages that are sent between {@link org.apache.flink.runtime.jobmaster.JobMaster}
+ * and {@link org.apache.flink.runtime.taskexecutor.TaskExecutor} to coordinate the checkpoint snapshots of the
  * distributed dataflow.
  */
-package org.apache.flink.runtime.messages.checkpoint;
+package org.apache.flink.runtime.messages.checkpoint;

flink-runtime/src/main/java/org/apache/flink/runtime/messages/package-info.java

@@ -17,8 +17,7 @@
  */
 
 /**
- * This package contains the messages that are sent between actors, like the
- * {@link org.apache.flink.runtime.jobmanager.JobManager} and
- * {@link org.apache.flink.runtime.taskmanager.TaskManager} to coordinate the distributed operations.
+ * This package contains the messages that are sent between Flink's distributed
+ * components to coordinate the distributed operations.
  */
-package org.apache.flink.runtime.messages;
+package org.apache.flink.runtime.messages;

flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateClientProxy.java

@@ -32,17 +32,17 @@
  *
  * <p>These are:
  * <ol>
- *     <li> the {@link org.apache.flink.runtime.jobmanager.JobManager Job Manager},
+ *     <li> the {@link org.apache.flink.runtime.jobmaster.JobMaster Job Manager},
  *     which is responsible for sending the
- *     {@link org.apache.flink.runtime.taskmanager.TaskManager Task Manager} storing
+ *     {@link org.apache.flink.runtime.taskexecutor.TaskExecutor Task Manager} storing
  *     the requested state, and </li>
  *     <li> the Task Manager having the state itself.</li>
  * </ol>
  */
 public interface KvStateClientProxy extends KvStateServer {
 
 	/**
-	 * Updates the active {@link org.apache.flink.runtime.jobmanager.JobManager Job Manager}
+	 * Updates the active {@link org.apache.flink.runtime.jobmaster.JobMaster Job Manager}
 	 * in case of change.
 	 *
 	 * <p>This is useful in settings where high-availability is enabled and

flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/history/ArchivedJson.java

@@ -18,7 +18,7 @@
 
 package org.apache.flink.runtime.webmonitor.history;
 
-import org.apache.flink.runtime.jobmanager.MemoryArchivist;
+import org.apache.flink.runtime.history.FsJobArchivist;
 import org.apache.flink.runtime.rest.messages.ResponseBody;
 import org.apache.flink.runtime.rest.util.RestMapperUtils;
 import org.apache.flink.util.Preconditions;
@@ -32,7 +32,7 @@
 /**
  * A simple container for a handler's JSON response and the REST URLs for which the response would've been returned.
  *
- * <p>These are created by {@link JsonArchivist}s, and used by the {@link MemoryArchivist} to create a directory structure
+ * <p>These are created by {@link JsonArchivist}s, and used by the {@link FsJobArchivist} to create a directory structure
  * resembling the REST API.
  */
 public class ArchivedJson {