/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.runtime.dispatcher;

import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
import org.apache.flink.runtime.entrypoint.JobClusterEntrypoint;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.util.FlinkException;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;

import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * Mini Dispatcher which is instantiated as the dispatcher component by the {@link JobClusterEntrypoint}.
 *
 * <p>The mini dispatcher is initialized with a single {@link JobGraph} which it runs.
 *
 * <p>Depending on the {@link ClusterEntrypoint.ExecutionMode}, the mini dispatcher will directly
 * terminate after job completion if its execution mode is {@link ClusterEntrypoint.ExecutionMode#DETACHED}.
 */
public class MiniDispatcher extends Dispatcher {

	private final JobClusterEntrypoint.ExecutionMode executionMode;

	public MiniDispatcher(
			RpcService rpcService,
			String endpointId,
			DispatcherId fencingToken,
			DispatcherServices dispatcherServices,
			JobGraph jobGraph,
			JobClusterEntrypoint.ExecutionMode executionMode) throws Exception {
		super(
			rpcService,
			endpointId,
			fencingToken,
			Collections.singleton(jobGraph),
			dispatcherServices);

		this.executionMode = checkNotNull(executionMode);
	}

	@Override
	public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout) {
		final CompletableFuture<Acknowledge> acknowledgeCompletableFuture = super.submitJob(jobGraph, timeout);

		acknowledgeCompletableFuture.whenComplete(
			(Acknowledge ignored, Throwable throwable) -> {
				if (throwable != null) {
					onFatalError(new FlinkException(
						"Failed to submit job " + jobGraph.getJobID() + " in job mode.",
						throwable));
				}
			});

		return acknowledgeCompletableFuture;
	}

	@Override
	public CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout) {
		final CompletableFuture<JobResult> jobResultFuture = super.requestJobResult(jobId, timeout);

		if (executionMode == ClusterEntrypoint.ExecutionMode.NORMAL) {
			// terminate the MiniDispatcher once we served the first JobResult successfully
			jobResultFuture.thenAccept((JobResult result) -> {
				ApplicationStatus status = result.getSerializedThrowable().isPresent() ?
						ApplicationStatus.FAILED : ApplicationStatus.SUCCEEDED;

				shutDownFuture.complete(status);
			});
		}

		return jobResultFuture;
	}

	@Override
	protected void jobReachedGloballyTerminalState(ArchivedExecutionGraph archivedExecutionGraph) {
		super.jobReachedGloballyTerminalState(archivedExecutionGraph);

		if (executionMode == ClusterEntrypoint.ExecutionMode.DETACHED) {
			// shut down since we don't have to wait for the execution result retrieval
			shutDownFuture.complete(ApplicationStatus.fromJobStatus(archivedExecutionGraph.getState()));
		}
	}

	@Override
	protected void jobNotFinished(JobID jobId) {
		super.jobNotFinished(jobId);

		// shut down since we have done our job
		shutDownFuture.complete(ApplicationStatus.UNKNOWN);
	}
}

@@ -1452,16 +1452,74 @@
 ption;%0A%0A
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0A
 import j
@@ -2115,16 +2115,98 @@
 atcher %7B
+%0A%09private static final Logger LOG = LoggerFactory.getLogger(MiniDispatcher.class);
 %0A%0A%09priva
@@ -3728,45 +3728,221 @@
 %09%09%09%09
-shutDownFuture.complete(status);%0A%09%09%09%7D
+LOG.debug(%22Shutting down cluster because someone retrieved the job result.%22);%0A%09%09%09%09shutDownFuture.complete(status);%0A%09%09%09%7D);%0A%09%09%7D else %7B%0A%09%09%09LOG.debug(%22Not shutting down cluster after someone retrieved the job result.%22
 );%0A%09

Add debug logging in MiniDispatcher