add stream sorter example for accumulator

Signed-off-by: Yashash H L <[email protected]>

Author: yhl25

examples/pom.xml

@@ -359,6 +359,28 @@
                             </to>
                         </configuration>
                     </execution>
+                    <execution>
+                        <id>accumulator-stream-sorter</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>dockerBuild</goal>
+                        </goals>
+                        <configuration>
+                            <from>
+                                <image>amazoncorretto:11</image>
+                            </from>
+                            <container>
+                                <mainClass>
+                                    io.numaproj.numaflow.examples.accumulator.sorter.StreamSorterFactory
+                                </mainClass>
+                            </container>
+                            <to>
+                                <image>
+                                    numaflow-java-examples/accumulator-stream-sorter:${docker.tag}
+                                </image>
+                            </to>
+                        </configuration>
+                    </execution>
                 </executions>
             </plugin>
         </plugins>

examples/src/main/java/io/numaproj/numaflow/examples/accumulator/sorter/StreamSorterFactory.java

@@ -0,0 +1,69 @@
+package io.numaproj.numaflow.examples.accumulator.sorter;
+
+import io.numaproj.numaflow.accumulator.Server;
+import io.numaproj.numaflow.accumulator.model.Accumulator;
+import io.numaproj.numaflow.accumulator.model.AccumulatorFactory;
+import io.numaproj.numaflow.accumulator.model.Datum;
+import io.numaproj.numaflow.accumulator.model.Message;
+import io.numaproj.numaflow.accumulator.model.OutputStreamObserver;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import java.time.Instant;
+import java.util.Comparator;
+import java.util.TreeSet;
+
+@Slf4j
+@AllArgsConstructor
+public class StreamSorterFactory extends AccumulatorFactory<StreamSorterFactory.StreamSorter> {
+
+    public static void main(String[] args) throws Exception {
+        log.info("Starting stream sorter server..");
+        Server server = new Server(new StreamSorterFactory());
+
+        // Start the server
+        server.start();
+
+        // wait for the server to shut down
+        server.awaitTermination();
+        log.info("Stream sorter server exited..");
+    }
+
+    @Override
+    public StreamSorter createAccumulator() {
+        return new StreamSorter();
+    }
+
+    public static class StreamSorter extends Accumulator {
+        private Instant latestWm = Instant.ofEpochMilli(-1);
+        private final TreeSet<Datum> sortedBuffer = new TreeSet<>(Comparator.comparing(Datum::getEventTime));
+
+        @Override
+        public void processMessage(Datum datum, OutputStreamObserver outputStream) {
+            log.info("Received datum with event time: {}", datum.toString());
+            if (datum.getWatermark().isAfter(latestWm)) {
+                latestWm = datum.getWatermark();
+                flushBuffer(outputStream);
+            }
+            sortedBuffer.add(datum);
+        }
+
+        @Override
+        public void handleEndOfStream(OutputStreamObserver outputStreamObserver) {
+            flushBuffer(outputStreamObserver);
+        }
+
+        private void flushBuffer(OutputStreamObserver outputStream) {
+            log.info("Watermark updated, flushing sortedBuffer: {}", latestWm.toEpochMilli());
+            while (!sortedBuffer.isEmpty() && !sortedBuffer
+                    .first()
+                    .getEventTime()
+                    .isAfter(latestWm)) {
+                Datum datum = sortedBuffer.pollFirst();
+                assert datum != null;
+                outputStream.send(new Message(datum));
+                log.info("Sent datum with event time: {}", datum.getEventTime().toEpochMilli());
+            }
+        }
+    }
+}

src/main/java/io/numaproj/numaflow/accumulator/Server.java

@@ -102,9 +102,7 @@ public void start() throws Exception {
      *         waiting
      */
     public void awaitTermination() throws InterruptedException {
-        log.info("accumulagtor server is waiting for termination");
         this.server.awaitTermination();
-        log.info("accumulagtor server terminated");
     }
 
     /**