Update StreamWorkerTaskRefresherTest.java

Merge conflict update: 1. Update BinlogEventListenerTest.java 2. Update BinlogEventListenerTest.java Signed-off-by: Dinu John <[email protected]>
dinujoh · Nov 12, 2024 · 0d4644f · 0d4644f
1 parent 8aa1d0e
commit 0d4644f
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 4 deletions.
diff --git a/...st/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListenerTest.java b/...st/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListenerTest.java
@@ -165,7 +165,7 @@ void test_given_DeleteRows_event_then_calls_correct_handler(EventType eventType)
     }
 
     private BinlogEventListener createObjectUnderTest() {
-        return new BinlogEventListener(buffer, sourceConfig, s3Prefix, pluginMetrics, binaryLogClient,
+        return BinlogEventListener.create(streamPartition, buffer, sourceConfig, s3Prefix, pluginMetrics, binaryLogClient,
                 streamCheckpointer, acknowledgementSetManager, dbTableMetadata, cascadingActionDetector);
     }
 
@@ -176,4 +176,4 @@ private void verifyHandlerCallHelper() {
         Runnable capturedRunnable = runnableArgumentCaptor.getValue();
         capturedRunnable.run();
     }
-}
+}
diff --git a/...a/org/opensearch/dataprepper/plugins/source/rds/stream/StreamWorkerTaskRefresherTest.java b/...a/org/opensearch/dataprepper/plugins/source/rds/stream/StreamWorkerTaskRefresherTest.java
@@ -125,7 +125,7 @@ void test_initialize_then_process_stream() {
             dbTableMetadataMockedStatic.when(() -> DbTableMetadata.fromMap(progressState)).thenReturn(dbTableMetadata);
             streamWorkerMockedStatic.when(() -> StreamWorker.create(eq(sourceCoordinator), any(BinaryLogClient.class), eq(pluginMetrics)))
                     .thenReturn(streamWorker);
-            binlogEventListenerMockedStatic.when(() -> BinlogEventListener.create(eq(buffer), any(RdsSourceConfig.class),
+            binlogEventListenerMockedStatic.when(() -> BinlogEventListener.create(eq(streamPartition), eq(buffer), any(RdsSourceConfig.class),
                             any(String.class), eq(pluginMetrics), eq(binlogClient), eq(streamCheckpointer),
                             eq(acknowledgementSetManager), eq(dbTableMetadata), any(CascadingActionDetector.class)))
                     .thenReturn(binlogEventListener);
@@ -163,7 +163,7 @@ void test_update_when_credentials_changed_then_refresh_task() {
             dbTableMetadataMockedStatic.when(() -> DbTableMetadata.fromMap(progressState)).thenReturn(dbTableMetadata);
             streamWorkerMockedStatic.when(() -> StreamWorker.create(eq(sourceCoordinator), any(BinaryLogClient.class), eq(pluginMetrics)))
                     .thenReturn(streamWorker);
-            binlogEventListenerMockedStatic.when(() -> BinlogEventListener.create(eq(buffer), any(RdsSourceConfig.class),
+            binlogEventListenerMockedStatic.when(() -> BinlogEventListener.create(eq(streamPartition), eq(buffer), any(RdsSourceConfig.class),
                             any(String.class), eq(pluginMetrics), eq(binlogClient), eq(streamCheckpointer),
                             eq(acknowledgementSetManager), eq(dbTableMetadata), any(CascadingActionDetector.class)))
                     .thenReturn(binlogEventListener);