[issue-717] Let Pravega internal thread pool deal with checkpoint (#718)

Signed-off-by: Brian Zhou <[email protected]>
pravega · Oct 11, 2023 · a55bfe1 · a55bfe1
1 parent 1157b7b
commit a55bfe1
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 7 deletions.
diff --git a/src/main/java/io/pravega/connectors/flink/ReaderCheckpointHook.java b/src/main/java/io/pravega/connectors/flink/ReaderCheckpointHook.java
@@ -114,7 +114,7 @@ public CompletableFuture<Checkpoint> triggerCheckpoint(
         final String checkpointName = createCheckpointName(checkpointId);
 
         final CompletableFuture<Checkpoint> checkpointResult =
-                this.readerGroup.initiateCheckpoint(checkpointName, scheduledExecutorService)
+                this.readerGroup.initiateCheckpoint(checkpointName)
                         .exceptionally(e -> {
                             if (e instanceof MaxNumberOfCheckpointsExceededException) {
                                 readerGroup.cancelOutstandingCheckpoints();

diff --git a/src/test/java/io/pravega/connectors/flink/ReaderCheckpointHookTest.java b/src/test/java/io/pravega/connectors/flink/ReaderCheckpointHookTest.java
@@ -66,10 +66,10 @@ public void testTriggerCheckpoint() throws Exception {
         CompletableFuture<Checkpoint> checkpointPromise = new CompletableFuture<>();
         TestableReaderCheckpointHook hook = new TestableReaderCheckpointHook(HOOK_UID, READER_GROUP_NAME, SCOPE, Time.minutes(1), clientConfig, readerGroupConfig);
 
-        when(hook.readerGroup.initiateCheckpoint(anyString(), any())).thenReturn(checkpointPromise);
+        when(hook.readerGroup.initiateCheckpoint(anyString())).thenReturn(checkpointPromise);
         CompletableFuture<Checkpoint> checkpointFuture = hook.triggerCheckpoint(1L, 1L, Executors.directExecutor());
         assertThat(checkpointFuture).isNotNull();
-        verify(hook.readerGroup).initiateCheckpoint(anyString(), any());
+        verify(hook.readerGroup).initiateCheckpoint(anyString());
 
         // complete the checkpoint promise
         Checkpoint expectedCheckpoint = mock(Checkpoint.class);
@@ -85,11 +85,11 @@ public void testTriggerCheckpointTimeout() throws Exception {
         CompletableFuture<Checkpoint> checkpointPromise = new CompletableFuture<>();
 
         TestableReaderCheckpointHook hook = new TestableReaderCheckpointHook(HOOK_UID, READER_GROUP_NAME, SCOPE, Time.minutes(1), clientConfig, readerGroupConfig);
-        when(hook.readerGroup.initiateCheckpoint(anyString(), any())).thenReturn(checkpointPromise);
+        when(hook.readerGroup.initiateCheckpoint(anyString())).thenReturn(checkpointPromise);
 
         CompletableFuture<Checkpoint> checkpointFuture = hook.triggerCheckpoint(1L, 1L, Executors.directExecutor());
         assertThat(checkpointFuture).isNotNull();
-        verify(hook.readerGroup).initiateCheckpoint(anyString(), any());
+        verify(hook.readerGroup).initiateCheckpoint(anyString());
 
         // invoke the timeout callback
         hook.invokeScheduledCallables();
@@ -104,11 +104,11 @@ public void testCancelWhenExceedingMaxOutstandingCheckpoints() throws Exception
         checkpointPromise.completeExceptionally(new MaxNumberOfCheckpointsExceededException("test"));
 
         TestableReaderCheckpointHook hook = new TestableReaderCheckpointHook(HOOK_UID, READER_GROUP_NAME, SCOPE, Time.minutes(1), clientConfig, readerGroupConfig);
-        when(hook.readerGroup.initiateCheckpoint(anyString(), any())).thenReturn(checkpointPromise);
+        when(hook.readerGroup.initiateCheckpoint(anyString())).thenReturn(checkpointPromise);
 
         CompletableFuture<Checkpoint> checkpointFuture = hook.triggerCheckpoint(1L, 1L, Executors.directExecutor());
         assertThat(checkpointFuture).isNotNull();
-        verify(hook.readerGroup).initiateCheckpoint(anyString(), any());
+        verify(hook.readerGroup).initiateCheckpoint(anyString());
 
         // invoke the cancelOutstandingCheckpoints
         verify(hook.readerGroup).cancelOutstandingCheckpoints();