Fix a bug with event listener in RDS source (#4962)

Unregister eventListener on stop Signed-off-by: Hai Yan <oeyh@amazon.com>
opensearch-project · Sep 20, 2024 · 18307d3 · 18307d3
1 parent aaef847
commit 18307d3
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 0 deletions.
diff --git a/...c/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListener.java b/...c/main/java/org/opensearch/dataprepper/plugins/source/rds/stream/BinlogEventListener.java
@@ -146,6 +146,7 @@ public void onEvent(com.github.shyiko.mysql.binlog.event.Event event) {
     public void stopClient() {
         try {
             binaryLogClient.disconnect();
+            binaryLogClient.unregisterEventListener(this);
             binlogEventExecutorService.shutdownNow();
             LOG.info("Binary log client disconnected.");
         } catch (Exception e) {

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
@@ -16,6 +16,7 @@
 import org.junit.jupiter.params.provider.EnumSource;
 import org.mockito.Answers;
 import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
 import org.mockito.Mock;
 import org.mockito.MockedStatic;
 import org.mockito.junit.jupiter.MockitoExtension;
@@ -26,6 +27,7 @@
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.plugins.source.rds.RdsSourceConfig;
 
+import java.io.IOException;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -34,6 +36,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mockStatic;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
@@ -103,6 +106,16 @@ void test_given_TableMap_event_then_calls_correct_handler() {
         verify(objectUnderTest).handleTableMapEvent(binlogEvent);
     }
 
+    @Test
+    void test_stopClient() throws IOException {
+        objectUnderTest.stopClient();
+
+        InOrder inOrder = inOrder(binaryLogClient, eventListnerExecutorService);
+        inOrder.verify(binaryLogClient).disconnect();
+        inOrder.verify(binaryLogClient).unregisterEventListener(objectUnderTest);
+        inOrder.verify(eventListnerExecutorService).shutdownNow();
+    }
+
     @ParameterizedTest
     @EnumSource(names = {"WRITE_ROWS", "EXT_WRITE_ROWS"})
     void test_given_WriteRows_event_then_calls_correct_handler(EventType eventType) {