SAMZA-2797: Call flush during stop from CoordinatorStreamWriter (#1692)

Author: ajothomas

samza-core/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.
    *

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();
   }