@Test public void testDisposeSavepoint() throws Exception { final String savepointPath = "foobar"; final String exceptionMessage = "Test exception."; final FlinkException testException = new FlinkException(exceptionMessage); final TestSavepointDisposalHandlers testSavepointDisposalHandlers = new TestSavepointDisposalHandlers(savepointPath); final TestSavepointDisposalHandlers.TestSavepointDisposalTriggerHandler testSavepointDisposalTriggerHandler = testSavepointDisposalHandlers.new TestSavepointDisposalTriggerHandler(); final TestSavepointDisposalHandlers.TestSavepointDisposalStatusHandler testSavepointDisposalStatusHandler = testSavepointDisposalHandlers.new TestSavepointDisposalStatusHandler( OptionalFailure.of(AsynchronousOperationInfo.complete()), OptionalFailure.of(AsynchronousOperationInfo.completeExceptional(new SerializedThrowable(testException))), OptionalFailure.ofFailure(testException)); try (TestRestServerEndpoint restServerEndpoint = createRestServerEndpoint( testSavepointDisposalStatusHandler, testSavepointDisposalTriggerHandler)) { RestClusterClient<?> restClusterClient = createRestClusterClient(restServerEndpoint.getServerAddress().getPort()); try { { final CompletableFuture<Acknowledge> disposeSavepointFuture = restClusterClient.disposeSavepoint(savepointPath); assertThat(disposeSavepointFuture.get(), is(Acknowledge.get())); } { final CompletableFuture<Acknowledge> disposeSavepointFuture = restClusterClient.disposeSavepoint(savepointPath); try { disposeSavepointFuture.get(); fail("Expected an exception"); } catch (ExecutionException ee) { assertThat(ExceptionUtils.findThrowableWithMessage(ee, exceptionMessage).isPresent(), is(true)); } } { try { restClusterClient.disposeSavepoint(savepointPath).get(); fail("Expected an exception."); } catch (ExecutionException ee) { assertThat(ExceptionUtils.findThrowable(ee, RestClientException.class).isPresent(), is(true)); } } } finally { restClusterClient.shutdown(); } } }
@Override public CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath) { final SavepointDisposalRequest savepointDisposalRequest = new SavepointDisposalRequest(savepointPath); final CompletableFuture<TriggerResponse> savepointDisposalTriggerFuture = sendRequest( SavepointDisposalTriggerHeaders.getInstance(), savepointDisposalRequest); final CompletableFuture<AsynchronousOperationInfo> savepointDisposalFuture = savepointDisposalTriggerFuture.thenCompose( (TriggerResponse triggerResponse) -> { final TriggerId triggerId = triggerResponse.getTriggerId(); final SavepointDisposalStatusHeaders savepointDisposalStatusHeaders = SavepointDisposalStatusHeaders.getInstance(); final SavepointDisposalStatusMessageParameters savepointDisposalStatusMessageParameters = savepointDisposalStatusHeaders.getUnresolvedMessageParameters(); savepointDisposalStatusMessageParameters.triggerIdPathParameter.resolve(triggerId); return pollResourceAsync( () -> sendRequest( savepointDisposalStatusHeaders, savepointDisposalStatusMessageParameters)); }); return savepointDisposalFuture.thenApply( (AsynchronousOperationInfo asynchronousOperationInfo) -> { if (asynchronousOperationInfo.getFailureCause() == null) { return Acknowledge.get(); } else { throw new CompletionException(asynchronousOperationInfo.getFailureCause()); } }); }
RestClusterClient extends ClusterClient<T> implements NewClusterClient { @Override public CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath) { final SavepointDisposalRequest savepointDisposalRequest = new SavepointDisposalRequest(savepointPath); final CompletableFuture<TriggerResponse> savepointDisposalTriggerFuture = sendRequest( SavepointDisposalTriggerHeaders.getInstance(), savepointDisposalRequest); final CompletableFuture<AsynchronousOperationInfo> savepointDisposalFuture = savepointDisposalTriggerFuture.thenCompose( (TriggerResponse triggerResponse) -> { final TriggerId triggerId = triggerResponse.getTriggerId(); final SavepointDisposalStatusHeaders savepointDisposalStatusHeaders = SavepointDisposalStatusHeaders.getInstance(); final SavepointDisposalStatusMessageParameters savepointDisposalStatusMessageParameters = savepointDisposalStatusHeaders.getUnresolvedMessageParameters(); savepointDisposalStatusMessageParameters.triggerIdPathParameter.resolve(triggerId); return pollResourceAsync( () -> sendRequest( savepointDisposalStatusHeaders, savepointDisposalStatusMessageParameters)); }); return savepointDisposalFuture.thenApply( (AsynchronousOperationInfo asynchronousOperationInfo) -> { if (asynchronousOperationInfo.getFailureCause() == null) { return Acknowledge.get(); } else { throw new CompletionException(asynchronousOperationInfo.getFailureCause()); } }); } }
RestClusterClient extends ClusterClient<T> implements NewClusterClient { @Override public CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath) { final SavepointDisposalRequest savepointDisposalRequest = new SavepointDisposalRequest(savepointPath); final CompletableFuture<TriggerResponse> savepointDisposalTriggerFuture = sendRequest( SavepointDisposalTriggerHeaders.getInstance(), savepointDisposalRequest); final CompletableFuture<AsynchronousOperationInfo> savepointDisposalFuture = savepointDisposalTriggerFuture.thenCompose( (TriggerResponse triggerResponse) -> { final TriggerId triggerId = triggerResponse.getTriggerId(); final SavepointDisposalStatusHeaders savepointDisposalStatusHeaders = SavepointDisposalStatusHeaders.getInstance(); final SavepointDisposalStatusMessageParameters savepointDisposalStatusMessageParameters = savepointDisposalStatusHeaders.getUnresolvedMessageParameters(); savepointDisposalStatusMessageParameters.triggerIdPathParameter.resolve(triggerId); return pollResourceAsync( () -> sendRequest( savepointDisposalStatusHeaders, savepointDisposalStatusMessageParameters)); }); return savepointDisposalFuture.thenApply( (AsynchronousOperationInfo asynchronousOperationInfo) -> { if (asynchronousOperationInfo.getFailureCause() == null) { return Acknowledge.get(); } else { throw new CompletionException(asynchronousOperationInfo.getFailureCause()); } }); } RestClusterClient(Configuration config, T clusterId);  RestClusterClient(
			Configuration config,
			T clusterId,
			LeaderRetrievalService webMonitorRetrievalService); @VisibleForTesting  RestClusterClient(
			Configuration configuration,
			@Nullable RestClient restClient,
			T clusterId,
			WaitStrategy waitStrategy,
			@Nullable LeaderRetrievalService webMonitorRetrievalService); }
RestClusterClient extends ClusterClient<T> implements NewClusterClient { @Override public CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath) { final SavepointDisposalRequest savepointDisposalRequest = new SavepointDisposalRequest(savepointPath); final CompletableFuture<TriggerResponse> savepointDisposalTriggerFuture = sendRequest( SavepointDisposalTriggerHeaders.getInstance(), savepointDisposalRequest); final CompletableFuture<AsynchronousOperationInfo> savepointDisposalFuture = savepointDisposalTriggerFuture.thenCompose( (TriggerResponse triggerResponse) -> { final TriggerId triggerId = triggerResponse.getTriggerId(); final SavepointDisposalStatusHeaders savepointDisposalStatusHeaders = SavepointDisposalStatusHeaders.getInstance(); final SavepointDisposalStatusMessageParameters savepointDisposalStatusMessageParameters = savepointDisposalStatusHeaders.getUnresolvedMessageParameters(); savepointDisposalStatusMessageParameters.triggerIdPathParameter.resolve(triggerId); return pollResourceAsync( () -> sendRequest( savepointDisposalStatusHeaders, savepointDisposalStatusMessageParameters)); }); return savepointDisposalFuture.thenApply( (AsynchronousOperationInfo asynchronousOperationInfo) -> { if (asynchronousOperationInfo.getFailureCause() == null) { return Acknowledge.get(); } else { throw new CompletionException(asynchronousOperationInfo.getFailureCause()); } }); } RestClusterClient(Configuration config, T clusterId);  RestClusterClient(
			Configuration config,
			T clusterId,
			LeaderRetrievalService webMonitorRetrievalService); @VisibleForTesting  RestClusterClient(
			Configuration configuration,
			@Nullable RestClient restClient,
			T clusterId,
			WaitStrategy waitStrategy,
			@Nullable LeaderRetrievalService webMonitorRetrievalService); @Override void shutdown(); @Override JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader); CompletableFuture<JobDetailsInfo> getJobDetails(JobID jobId); @Override CompletableFuture<JobStatus> getJobStatus(JobID jobId); @Override CompletableFuture<JobResult> requestJobResult(@Nonnull JobID jobId); @Override CompletableFuture<JobSubmissionResult> submitJob(@Nonnull JobGraph jobGraph); @Override void stop(JobID jobID); @Override void cancel(JobID jobID); @Override String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirectory); @Override CompletableFuture<String> triggerSavepoint(
			final JobID jobId,
			final @Nullable String savepointDirectory); @Override Map<String, OptionalFailure<Object>> getAccumulators(final JobID jobID, ClassLoader loader); @Override CompletableFuture<Collection<JobStatusMessage>> listJobs(); @Override T getClusterId(); @Override LeaderConnectionInfo getClusterConnectionInfo(); @Override CompletableFuture<Acknowledge> rescaleJob(JobID jobId, int newParallelism); @Override CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath); @Override void shutDownCluster(); @Override boolean hasUserJarsInClassPath(List<URL> userJarFiles); @Override void waitForClusterToBeReady(); @Override String getWebInterfaceURL(); @Override GetClusterStatusResponse getClusterStatus(); @Override List<String> getNewMessages(); @Override int getMaxSlots(); @VisibleForTesting CompletableFuture<P> sendRequest(M messageHeaders, U messageParameters, R request); }
RestClusterClient extends ClusterClient<T> implements NewClusterClient { @Override public CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath) { final SavepointDisposalRequest savepointDisposalRequest = new SavepointDisposalRequest(savepointPath); final CompletableFuture<TriggerResponse> savepointDisposalTriggerFuture = sendRequest( SavepointDisposalTriggerHeaders.getInstance(), savepointDisposalRequest); final CompletableFuture<AsynchronousOperationInfo> savepointDisposalFuture = savepointDisposalTriggerFuture.thenCompose( (TriggerResponse triggerResponse) -> { final TriggerId triggerId = triggerResponse.getTriggerId(); final SavepointDisposalStatusHeaders savepointDisposalStatusHeaders = SavepointDisposalStatusHeaders.getInstance(); final SavepointDisposalStatusMessageParameters savepointDisposalStatusMessageParameters = savepointDisposalStatusHeaders.getUnresolvedMessageParameters(); savepointDisposalStatusMessageParameters.triggerIdPathParameter.resolve(triggerId); return pollResourceAsync( () -> sendRequest( savepointDisposalStatusHeaders, savepointDisposalStatusMessageParameters)); }); return savepointDisposalFuture.thenApply( (AsynchronousOperationInfo asynchronousOperationInfo) -> { if (asynchronousOperationInfo.getFailureCause() == null) { return Acknowledge.get(); } else { throw new CompletionException(asynchronousOperationInfo.getFailureCause()); } }); } RestClusterClient(Configuration config, T clusterId);  RestClusterClient(
			Configuration config,
			T clusterId,
			LeaderRetrievalService webMonitorRetrievalService); @VisibleForTesting  RestClusterClient(
			Configuration configuration,
			@Nullable RestClient restClient,
			T clusterId,
			WaitStrategy waitStrategy,
			@Nullable LeaderRetrievalService webMonitorRetrievalService); @Override void shutdown(); @Override JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader); CompletableFuture<JobDetailsInfo> getJobDetails(JobID jobId); @Override CompletableFuture<JobStatus> getJobStatus(JobID jobId); @Override CompletableFuture<JobResult> requestJobResult(@Nonnull JobID jobId); @Override CompletableFuture<JobSubmissionResult> submitJob(@Nonnull JobGraph jobGraph); @Override void stop(JobID jobID); @Override void cancel(JobID jobID); @Override String cancelWithSavepoint(JobID jobId, @Nullable String savepointDirectory); @Override CompletableFuture<String> triggerSavepoint(
			final JobID jobId,
			final @Nullable String savepointDirectory); @Override Map<String, OptionalFailure<Object>> getAccumulators(final JobID jobID, ClassLoader loader); @Override CompletableFuture<Collection<JobStatusMessage>> listJobs(); @Override T getClusterId(); @Override LeaderConnectionInfo getClusterConnectionInfo(); @Override CompletableFuture<Acknowledge> rescaleJob(JobID jobId, int newParallelism); @Override CompletableFuture<Acknowledge> disposeSavepoint(String savepointPath); @Override void shutDownCluster(); @Override boolean hasUserJarsInClassPath(List<URL> userJarFiles); @Override void waitForClusterToBeReady(); @Override String getWebInterfaceURL(); @Override GetClusterStatusResponse getClusterStatus(); @Override List<String> getNewMessages(); @Override int getMaxSlots(); @VisibleForTesting CompletableFuture<P> sendRequest(M messageHeaders, U messageParameters, R request);  }