SAMZA-2797: Call flush during stop from CoordinatorStreamWriter (apac…

…he#1692)
mynameborat · Nov 22, 2023 · e1816f3 · e1816f3
1 parent 66495b6
commit e1816f3
Show file tree

Hide file tree

Showing 2 changed files with 9 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
@@ -110,6 +110,14 @@ public void stop() {
     isStarted = false;
   }
 
+  /**
+   * Flushes underlying system producer.
+   * */
+  public void flush(String source) {
+    log.info("Flushing coordinator stream producer.");
+    systemProducer.flush(source);
+  }
+
   /**
    * Serialize and send a coordinator stream message.
    *

diff --git a/samza-core/src/main/java/org/apache/samza/coordinator/stream/CoordinatorStreamWriter.java b/samza-core/src/main/java/org/apache/samza/coordinator/stream/CoordinatorStreamWriter.java
@@ -65,6 +65,7 @@ public void start() {
    */
   public void stop() {
     log.info("Stopping the coordinator stream producer.");
+    coordinatorStreamSystemProducer.flush(SOURCE);
     coordinatorStreamSystemProducer.stop();
   }