### Question:
ClusterClient { public void shutdown() throws Exception { synchronized (this) { actorSystemLoader.shutdown(); if (!sharedHaServices && highAvailabilityServices != null) { highAvailabilityServices.close(); } } } ClusterClient(Configuration flinkConfig);  ClusterClient(Configuration flinkConfig, HighAvailabilityServices highAvailabilityServices, boolean sharedHaServices); void shutdown(); void setPrintStatusDuringExecution(boolean print); boolean getPrintStatusDuringExecution(); LeaderConnectionInfo getClusterConnectionInfo(); static String getOptimizedPlanAsJson(Optimizer compiler, PackagedProgram prog, int parallelism); static FlinkPlan getOptimizedPlan(Optimizer compiler, PackagedProgram prog, int parallelism); static OptimizedPlan getOptimizedPlan(Optimizer compiler, Plan p, int parallelism); JobSubmissionResult run(PackagedProgram prog, int parallelism); JobSubmissionResult run(JobWithJars program, int parallelism); JobSubmissionResult run(JobWithJars jobWithJars, int parallelism, SavepointRestoreSettings savepointSettings); JobSubmissionResult run(
			FlinkPlan compiledPlan, List<URL> libraries, List<URL> classpaths, ClassLoader classLoader); JobSubmissionResult run(FlinkPlan compiledPlan,
			List<URL> libraries, List<URL> classpaths, ClassLoader classLoader, SavepointRestoreSettings savepointSettings); JobExecutionResult run(JobGraph jobGraph, ClassLoader classLoader); JobSubmissionResult runDetached(JobGraph jobGraph, ClassLoader classLoader); JobExecutionResult retrieveJob(JobID jobID); JobListeningContext connectToJob(JobID jobID); CompletableFuture<JobStatus> getJobStatus(JobID jobId); void cancel(JobID jobId); String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirectory); void stop(final JobID jobId); CompletableFuture<String> triggerSavepoint(JobID jobId, @Nullable String savepointDirectory); CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath); CompletableFuture<Collection<JobStatusMessage>> listJobs(); Map<String, OptionalFailure<Object>> getAccumulators(JobID jobID); Map<String, OptionalFailure<Object>> getAccumulators(JobID jobID, ClassLoader loader); void endSession(JobID jobId); void endSessions(List<JobID> jobIds); static JobGraph getJobGraph(Configuration flinkConfig, PackagedProgram prog, FlinkPlan optPlan, SavepointRestoreSettings savepointSettings); static JobGraph getJobGraph(Configuration flinkConfig, FlinkPlan optPlan, List<URL> jarFiles, List<URL> classpaths, SavepointRestoreSettings savepointSettings); ActorGateway getJobManagerGateway(); abstract void waitForClusterToBeReady(); abstract String getWebInterfaceURL(); abstract GetClusterStatusResponse getClusterStatus(); abstract List<String> getNewMessages(); abstract T getClusterId(); void setDetached(boolean isDetached); boolean isDetached(); Configuration getFlinkConfiguration(); abstract int getMaxSlots(); abstract boolean hasUserJarsInClassPath(List<URL> userJarFiles); abstract JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader); CompletableFuture<Acknowledge> rescaleJob(JobID jobId, int newParallelism); void shutDownCluster(); static final int MAX_SLOTS_UNKNOWN; }### Answer:
@Test public void testClusterClientShutdown() throws Exception { Configuration config = new Configuration(); HighAvailabilityServices highAvailabilityServices = mock(HighAvailabilityServices.class); StandaloneClusterClient clusterClient = new StandaloneClusterClient(config, highAvailabilityServices, false); clusterClient.shutdown(); verify(highAvailabilityServices, never()).closeAndCleanupAllData(); verify(highAvailabilityServices).close(); }