* Provides options to retrieve cluster information in both application and session modes.
 * <p>
 * This interface offers a standardized way to access cluster-specific configuration options,
 * which can be used to generate a unique cluster ID.
 * </p>
 *
 */
public interface ClusterInfo {

    /**
     * Retrieves the options necessary for cluster information.
     * <p>
     * These options can be used to generate a cluster ID using the
     * {@code org.apache.flink.client.deployment.ClusterClientFactory#getClusterId(Configuration)} method.
     * </p>
     *
     * @return A map containing key-value pairs of cluster information options.
     */
    Map<String, String> getOptions();

    /**
     * Creates a ClusterInfo instance with a single option.
     *
     * @param clusterIdKey The key for the cluster ID option.
     * @param clusterIdValue The value for the cluster ID option.
     * @return A new ClusterInfo instance containing the specified option.
     */
    static ClusterInfo of(String clusterIdKey, String clusterIdValue) {
        return new ClusterInfoImpl(clusterIdKey, clusterIdValue);
    }

    /**
     * Creates an empty ClusterInfo instance.
     *
     * @return A ClusterInfo instance with no options.
     */
    static ClusterInfo empty() {
        return Collections::emptyMap;
    }
}     

@PublicEvolving
public interface JobClient {

    /** Returns the {@link JobID} that uniquely identifies the job this client is scoped to. */
    JobID getJobID();

    /** Requests the {@link JobStatus} of the associated job. */
    CompletableFuture<JobStatus> getJobStatus();

    /** Cancels the associated job. */
    CompletableFuture<Void> cancel();

    default ClusterInfo getClusterInfo() {
        return ClusterInfo.empty();
    }
}

/**
 * Implementation of the ClusterInfo interface that stores a single cluster ID option.
 * <p>
 * This class provides a simple way to create a ClusterInfo with one key-value pair
 * representing the cluster ID option.
 * </p>
 */
public class ClusterInfoImpl implements ClusterInfo {

    /** The key for the cluster ID option. */
    private final String clusterIdKey;

    /** The value for the cluster ID option. */
    private final String clusterIdValue;

    /**
     * Constructs a new ClusterInfoImpl with the specified cluster ID option.
     *
     * @param clusterIdKey The key for the cluster ID option.
     * @param clusterIdValue The value for the cluster ID option.
     */
    public ClusterInfoImpl(String clusterIdKey, String clusterIdValue) {
        this.clusterIdKey = clusterIdKey;
        this.clusterIdValue = clusterIdValue;
    }

    /**
     * Returns a map containing the single cluster ID option.
     *
     * @return A map with one entry, where the key is the cluster ID option key
     *         and the value is the cluster ID option value.
     */
    @Override
    public Map<String, String> getOptions() {
        return Collections.singletonMap(clusterIdKey, clusterIdValue);
    }
}

public class ClusterClientJobClientAdapter<ClusterID>
        implements JobClient, CoordinationRequestGateway {

    private final ClusterClientProvider<ClusterID> clusterClientProvider;

    private final JobID jobID;

    private final ClassLoader classLoader;

    private final ClusterInfo clusterInfo;

    public ClusterClientJobClientAdapter(
            final ClusterClientProvider<ClusterID> clusterClientProvider,
            final JobID jobID,
            final ClassLoader classLoader,
            final ClusterInfo clusterInfo) {
        this.jobID = checkNotNull(jobID);
        this.clusterClientProvider = checkNotNull(clusterClientProvider);
        this.classLoader = classLoader;
        this.clusterInfo = clusterInfo;
    }
    
    @Override
    public ClusterInfo getClusterInfo() {
        return clusterInfo;
    }

    ...
}

public class KubernetesClusterClientFactory
       extends AbstractContainerizedClusterClientFactory<String> {

    @Override
    public ClusterInfo getClusterInfo(String clusterID) {
        return ClusterInfo.of(KubernetesConfigOptions.CLUSTER_ID.key(), clusterID);
    }
}


public class YarnClusterClientFactory
       extends AbstractContainerizedClusterClientFactory<ApplicationId> {

    @Override
    public ClusterInfo getClusterInfo(ApplicationId clusterID) {
        return ClusterInfo.of(YarnConfigOptions.APPLICATION_ID.key(), clusterID.toString());
    }
}