[FLINK-11389] Incorrectly use job information when call getSerializedTaskInformation in class TaskDeploymentDescriptor

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.6.3, 1.7.1, 1.8.0
• Fix Version/s: 1.6.4, 1.7.2, 1.8.0
• Component/s: Runtime / Coordination
• Labels: pull-request-available
• Assignee: yuqi
• Reporter: yuqi
• Created: 18/Jan/19 08:23
• Updated: 31/Jan/19 14:51
• Resolved: 31/Jan/19 14:51

Description

See TaskDeploymentDescriptor

@Nullable
	public SerializedValue<TaskInformation> getSerializedTaskInformation() {
		if (serializedJobInformation instanceof NonOffloaded) {
			NonOffloaded<TaskInformation> jobInformation =
				(NonOffloaded<TaskInformation>) serializedTaskInformation;
			return jobInformation.serializedValue;
		} else {
			throw new IllegalStateException(
				"Trying to work with offloaded serialized job information.");
		}
	}

the condition serializedJobInformation instanceof NonOffloaded is not correctly,
as serializedJobInformation and serializedTaskInformation are passed from ExecutionVertex#createDeploymentDescriptor

		if (jobInformationOrBlobKey.isLeft()) {
			serializedJobInformation = new TaskDeploymentDescriptor.NonOffloaded<>(jobInformationOrBlobKey.left());
		} else {
			serializedJobInformation = new TaskDeploymentDescriptor.Offloaded<>(jobInformationOrBlobKey.right());
		}

		final Either<SerializedValue<TaskInformation>, PermanentBlobKey> taskInformationOrBlobKey;

		try {
			taskInformationOrBlobKey = jobVertex.getTaskInformationOrBlobKey();
		} catch (IOException e) {
			throw new ExecutionGraphException(
				"Could not create a serialized JobVertexInformation for " +
					jobVertex.getJobVertexId(), e);
		}

		final TaskDeploymentDescriptor.MaybeOffloaded<TaskInformation> serializedTaskInformation;

		if (taskInformationOrBlobKey.isLeft()) {
			serializedTaskInformation = new TaskDeploymentDescriptor.NonOffloaded<>(taskInformationOrBlobKey.left());
		} else {
			serializedTaskInformation = new TaskDeploymentDescriptor.Offloaded<>(taskInformationOrBlobKey.right());
		}

serializedJobInformation and serializedTaskInformation are not necessarily shared the class NonOffloaded or Offloaded

Issue Links

• links to: GitHub Pull Request #7532