feat: Test Kit for Servers (#111)

Signed-off-by: Yashash H L <[email protected]> Signed-off-by: Keran Yang <[email protected]> Co-authored-by: Keran Yang <[email protected]>
numaproj · Apr 15, 2024 · 0ac0102 · 0ac0102
1 parent ca13ac1
commit 0ac0102
Show file tree

Hide file tree

Showing 91 changed files with 2,288 additions and 366 deletions.
diff --git a/README.md b/README.md
@@ -38,8 +38,6 @@ Add this dependency to your project's build file:
 compile "io.numaproj.numaflow:numaflow-java:0.7.0"
 ```
 
-```
-
 ### Build
 
 ```bash

diff --git a/examples/pom.xml b/examples/pom.xml
@@ -25,16 +25,18 @@
             <artifactId>slf4j-simple</artifactId>
             <version>2.0.7</version>
         </dependency>
+
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
             <version>4.8.1</version>
             <scope>test</scope>
         </dependency>
+
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter</artifactId>
-            <version>RELEASE</version>
+            <version>5.10.2</version>
             <scope>test</scope>
         </dependency>
     </dependencies>
@@ -59,7 +61,9 @@
                                 </mainClass>
                             </container>
                             <to>
-                                <image>numaflow-java-examples/mapt-event-time-filter-function:${docker.tag}</image>
+                                <image>
+                                    numaflow-java-examples/mapt-event-time-filter-function:${docker.tag}
+                                </image>
                             </to>
                         </configuration>
                     </execution>
@@ -160,7 +164,8 @@
                                 </mainClass>
                             </container>
                             <to>
-                                <image>numaflow-java-examples/reduce-stream-sum:${docker.tag}</image>
+                                <image>numaflow-java-examples/reduce-stream-sum:${docker.tag}
+                                </image>
                             </to>
                         </configuration>
                     </execution>
@@ -177,7 +182,8 @@
                                 </mainClass>
                             </container>
                             <to>
-                                <image>numaflow-java-examples/map-forward-message:${docker.tag}</image>
+                                <image>numaflow-java-examples/map-forward-message:${docker.tag}
+                                </image>
                             </to>
                         </configuration>
                     </execution>
@@ -211,7 +217,8 @@
                                 </mainClass>
                             </container>
                             <to>
-                                <image>numaflow-java-examples/sideinput-example:${docker.tag}</image>
+                                <image>numaflow-java-examples/sideinput-example:${docker.tag}
+                                </image>
                             </to>
                         </configuration>
                     </execution>
@@ -228,7 +235,8 @@
                                 </mainClass>
                             </container>
                             <to>
-                                <image>numaflow-java-examples/udf-sideinput-example:${docker.tag}</image>
+                                <image>numaflow-java-examples/udf-sideinput-example:${docker.tag}
+                                </image>
                             </to>
                         </configuration>
                     </execution>
@@ -245,7 +253,8 @@
                                 </mainClass>
                             </container>
                             <to>
-                                <image>numaflow-java-examples/source-simple-source:${docker.tag}</image>
+                                <image>numaflow-java-examples/source-simple-source:${docker.tag}
+                                </image>
                             </to>
                         </configuration>
                     </execution>
@@ -262,7 +271,8 @@
                                 </mainClass>
                             </container>
                             <to>
-                                <image>numaflow-java-examples/session-reduce-count:${docker.tag}</image>
+                                <image>numaflow-java-examples/session-reduce-count:${docker.tag}
+                                </image>
                             </to>
                         </configuration>
                     </execution>
@@ -271,6 +281,10 @@
         </plugins>
         <pluginManagement>
             <plugins>
+                <plugin>
+                    <artifactId>maven-surefire-plugin</artifactId>
+                    <version>2.22.2</version>
+                </plugin>
                 <plugin>
                     <artifactId>maven-compiler-plugin</artifactId>
                     <version>3.10.1</version>

diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/map/flatmap/FlatMapFunction.java b/examples/src/main/java/io/numaproj/numaflow/examples/map/flatmap/FlatMapFunction.java
@@ -16,7 +16,13 @@
 public class FlatMapFunction extends Mapper {
 
     public static void main(String[] args) throws Exception {
-        new Server(new FlatMapFunction()).start();
+        Server server = new Server(new FlatMapFunction());
+
+        // Start the server
+        server.start();
+
+        // Wait for the server to shut down
+        server.awaitTermination();
     }
 
     public MessageList processMessage(String[] keys, Datum data) {

diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/map/forward/ForwardFunction.java b/examples/src/main/java/io/numaproj/numaflow/examples/map/forward/ForwardFunction.java
@@ -12,7 +12,13 @@
 
 public class ForwardFunction extends Mapper {
     public static void main(String[] args) throws Exception {
-        new Server(new ForwardFunction()).start();
+        Server server = new Server(new ForwardFunction());
+
+        // Start the server
+        server.start();
+
+        // Wait for the server to shut down
+        server.awaitTermination();
     }
 
     public MessageList processMessage(String[] keys, Datum data) {

diff --git a/...ain/java/io/numaproj/numaflow/examples/mapstream/flatmapstream/FlatMapStreamFunction.java b/...ain/java/io/numaproj/numaflow/examples/mapstream/flatmapstream/FlatMapStreamFunction.java
@@ -1,10 +1,6 @@
 package io.numaproj.numaflow.examples.mapstream.flatmapstream;
 
-import io.numaproj.numaflow.mapstreamer.Datum;
-import io.numaproj.numaflow.mapstreamer.MapStreamer;
-import io.numaproj.numaflow.mapstreamer.Message;
-import io.numaproj.numaflow.mapstreamer.OutputObserver;
-import io.numaproj.numaflow.mapstreamer.Server;
+import io.numaproj.numaflow.mapstreamer.*;
 
 
 /**
@@ -17,7 +13,13 @@
 public class FlatMapStreamFunction extends MapStreamer {
 
     public static void main(String[] args) throws Exception {
-        new Server(new FlatMapStreamFunction()).start();
+        Server server = new Server(new FlatMapStreamFunction());
+
+        // Start the server
+        server.start();
+
+        // wait for the server to shutdown
+        server.awaitTermination();
     }
 
     public void processMessage(String[] keys, Datum data, OutputObserver outputObserver) {

diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/reduce/count/CounterFactory.java b/examples/src/main/java/io/numaproj/numaflow/examples/reduce/count/CounterFactory.java
@@ -23,7 +23,13 @@ public class CounterFactory extends ReducerFactory<CounterFactory.Counter> {
     public static void main(String[] args) throws Exception {
         log.info("counter udf was invoked");
         Config config = new Config(1);
-        new Server(new CounterFactory(config)).start();
+        Server server = new Server(new CounterFactory(config));
+
+        // Start the server
+        server.start();
+
+        // wait for the server to shut down
+        server.awaitTermination();
     }
 
     @Override
@@ -42,8 +48,8 @@ public Counter(Config config) {
 
         @Override
         public void addMessage(String[] keys, Datum datum, Metadata md) {
-                // increment based on the value specified in the config
-                count += config.getIncrementBy();
+            // increment based on the value specified in the config
+            count += config.getIncrementBy();
         }
 
         @Override

diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/reduce/sum/SumFactory.java b/examples/src/main/java/io/numaproj/numaflow/examples/reduce/sum/SumFactory.java
@@ -9,7 +9,13 @@ public class SumFactory extends ReducerFactory<SumFunction> {
 
     public static void main(String[] args) throws Exception {
         log.info("sum udf was invoked");
-        new Server(new SumFactory()).start();
+        Server server = new Server(new SumFactory());
+
+        // Start the server
+        server.start();
+
+        // wait for the server to shut down
+        server.awaitTermination();
     }
 
     @Override

diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/reducesession/counter/CountFactory.java b/examples/src/main/java/io/numaproj/numaflow/examples/reducesession/counter/CountFactory.java
@@ -13,7 +13,13 @@ public class CountFactory extends SessionReducerFactory<CountFunction> {
 
     public static void main(String[] args) throws Exception {
         log.info("count udf was invoked");
-        new Server(new CountFactory()).start();
+        Server server = new Server(new CountFactory());
+
+        // Start the server
+        server.start();
+
+        // wait for the server to shut down
+        server.awaitTermination();
     }
 
     @Override

diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/reducestreamer/sum/SumFactory.java b/examples/src/main/java/io/numaproj/numaflow/examples/reducestreamer/sum/SumFactory.java
@@ -13,7 +13,13 @@ public class SumFactory extends ReduceStreamerFactory<SumFunction> {
 
     public static void main(String[] args) throws Exception {
         log.info("sum udf was invoked");
-        new Server(new SumFactory()).start();
+        Server server = new Server(new SumFactory());
+
+        // Start the server
+        server.start();
+
+        // wait for the server to shut down
+        server.awaitTermination();
     }
 
     @Override

diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/sideinput/simple/SimpleSideInput.java b/examples/src/main/java/io/numaproj/numaflow/examples/sideinput/simple/SimpleSideInput.java
@@ -28,7 +28,13 @@ public SimpleSideInput(Config config) {
     }
 
     public static void main(String[] args) throws Exception {
-        new Server(new SimpleSideInput(new Config("sampling", 0.5F))).start();
+        Server server = new Server(new SimpleSideInput(new Config("sampling", 0.5F)));
+
+        // Start the server
+        server.start();
+
+        // wait for the server to shut down
+        server.awaitTermination();
     }
 
     @Override

diff --git a/...les/src/main/java/io/numaproj/numaflow/examples/sideinput/udf/SimpleMapWithSideInput.java b/...les/src/main/java/io/numaproj/numaflow/examples/sideinput/udf/SimpleMapWithSideInput.java
@@ -38,7 +38,13 @@ public static void main(String[] args) throws Exception {
         sideInputWatcher.startWatching();
 
         // start the server
-        new Server(new SimpleMapWithSideInput(sideInputWatcher)).start();
+        Server server = new Server(new SimpleMapWithSideInput(sideInputWatcher));
+
+        // Start the server
+        server.start();
+
+        // wait for the server to shut down
+        server.awaitTermination();
 
         // Stop watching for side input
         sideInputWatcher.stopWatching();

diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/sink/simple/SimpleSink.java b/examples/src/main/java/io/numaproj/numaflow/examples/sink/simple/SimpleSink.java
@@ -17,7 +17,13 @@
 public class SimpleSink extends Sinker {
 
     public static void main(String[] args) throws Exception {
-        new Server(new SimpleSink()).start();
+        Server server = new Server(new SimpleSink());
+
+        // Start the server
+        server.start();
+
+        // wait for the server to shut down
+        server.awaitTermination();
     }
 
     @Override

diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/source/simple/SimpleSource.java b/examples/src/main/java/io/numaproj/numaflow/examples/source/simple/SimpleSource.java
@@ -1,6 +1,5 @@
 package io.numaproj.numaflow.examples.source.simple;
 
-import com.google.common.annotations.VisibleForTesting;
 import com.google.common.primitives.Longs;
 import io.numaproj.numaflow.sourcer.AckRequest;
 import io.numaproj.numaflow.sourcer.Message;
@@ -29,7 +28,13 @@ public class SimpleSource extends Sourcer {
     private long readIndex = 0;
 
     public static void main(String[] args) throws Exception {
-        new Server(new SimpleSource()).start();
+        Server server = new Server(new SimpleSource());
+
+        // Start the server
+        server.start();
+
+        // wait for the server to shut down
+        server.awaitTermination();
     }
 
     @Override

diff --git a/...numaproj/numaflow/examples/sourcetransformer/eventtimefilter/EventTimeFilterFunction.java b/...numaproj/numaflow/examples/sourcetransformer/eventtimefilter/EventTimeFilterFunction.java
@@ -23,8 +23,13 @@ public class EventTimeFilterFunction extends SourceTransformer {
     private static final Instant januaryFirst2023 = Instant.ofEpochMilli(1672531200000L);
 
     public static void main(String[] args) throws Exception {
-        new Server(new EventTimeFilterFunction())
-                .start();
+        Server server = new Server(new EventTimeFilterFunction());
+
+        // Start the server
+        server.start();
+
+        // wait for the server to shut down
+        server.awaitTermination();
     }
 
     public MessageList processMessage(String[] keys, Datum data) {