SAMZA-2797: Add flush to CoordinatorStreamSystemProducer stop to ensu…

…re async sends are flushed prior to stop
apache · Nov 21, 2023 · 5c2df1c · 5c2df1c
1 parent 66495b6
commit 5c2df1c
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/...re/src/main/java/org/apache/samza/coordinator/stream/CoordinatorStreamSystemProducer.java b/...re/src/main/java/org/apache/samza/coordinator/stream/CoordinatorStreamSystemProducer.java
@@ -40,6 +40,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.samza.coordinator.stream.CoordinatorStreamWriter.SOURCE;
+
 /**
  * A wrapper around a SystemProducer that provides helpful methods for dealing
  * with the coordinator stream.
@@ -105,6 +107,7 @@ public void start() {
    */
   public void stop() {
     log.info("Stopping coordinator stream producer.");
+    systemProducer.flush(SOURCE);
     systemProducer.stop();
     systemAdmin.stop();
     isStarted = false;