Close mutation sender after completing import (#141)

datastax · Apr 11, 2023 · 1b37ff8 · 1b37ff8
1 parent cdeda9c
commit 1b37ff8
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 0 deletions.
diff --git a/backfill-cli/src/main/java/com/datastax/oss/cdc/backfill/importer/PulsarImporter.java b/backfill-cli/src/main/java/com/datastax/oss/cdc/backfill/importer/PulsarImporter.java
@@ -210,6 +210,13 @@ public ExitStatus importTable() {
                     LOGGER.warn("Error while closing CVS connector", e);
                 }
             }
+            if (mutationSender != null) {
+                try {
+                    mutationSender.close();
+                } catch (Exception e) {
+                    LOGGER.warn("Error while closing Pulsar mutation sender", e);
+                }
+            }
             printSummary(recordsCount);
         }
     }

diff --git a/backfill-cli/src/test/java/com/datastax/oss/cdc/backfill/PulsarImporterTest.java b/backfill-cli/src/test/java/com/datastax/oss/cdc/backfill/PulsarImporterTest.java
@@ -128,6 +128,7 @@ public void testImportPartitionKeyOnly() {
         // then
         assertEquals(ExitStatus.STATUS_OK, status);
         Mockito.verify(sender, Mockito.times(2)).sendMutationAsync(abstractMutationCaptor.capture());
+        Mockito.verify(sender, Mockito.times(1)).close();
         List<AbstractMutation<TableMetadata>> pkValues = abstractMutationCaptor.getAllValues();
         assertEquals(2, pkValues.size());
         List<Object> allPkValues = pkValues.stream().flatMap(v-> Arrays.stream(v.getPkValues())).collect(Collectors.toList());
@@ -203,6 +204,7 @@ public void testImportPartitionAndClusteringKeys() {
         // then
         assertEquals(ExitStatus.STATUS_OK, status);
         Mockito.verify(sender, Mockito.times(2)).sendMutationAsync(abstractMutationCaptor.capture());
+        Mockito.verify(sender, Mockito.times(1)).close();
         List<AbstractMutation<TableMetadata>> pkValues = abstractMutationCaptor.getAllValues();
         assertEquals(2, pkValues.size());
         List<Object>[] allPkValues = pkValues.stream().map(v-> v.getPkValues()).map(Arrays::asList).toArray(List[]::new);
@@ -285,6 +287,7 @@ public void testImportInflightMessagesBound() throws URISyntaxException, IOExcep
         // verify that no more interactions with sender because no new records should've been sent.
         assertTrue(importFuture.isDone());
         assertThat(importFuture.get(), is(ExitStatus.STATUS_OK));
+        Mockito.verify(sender, Mockito.times(1)).close();
         Mockito.verifyNoMoreInteractions(sender);
     }
 
@@ -330,6 +333,7 @@ public void testImportFailsFast() throws URISyntaxException, IOException, Execut
 
         // then
         assertTrue(importFuture.isDone());
+        Mockito.verify(sender, Mockito.times(1)).close();
         assertThat(importFuture.get(), is(ExitStatus.STATUS_ABORTED_FATAL_ERROR));
     }