chore: update sink interface (#69)

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

Author: yhl25

README.md

@@ -25,8 +25,7 @@ Add this dependency to your project's POM:
 <dependency>
   <groupId>io.numaproj.numaflow</groupId>
   <artifactId>numaflow-java</artifactId>
-  <version>${latest}</version>
-  <scope>compile</scope>
+  <version>0.5.1</version>
 </dependency>
 ```
 
@@ -35,7 +34,8 @@ Add this dependency to your project's POM:
 Add this dependency to your project's build file:
 
 ```groovy
-compile "io.numaproj.numaflow:numaflow-java:${latest}"
+compile "io.numaproj.numaflow:numaflow-java:0.5.1"
+```
 ```
 
 ### Build

examples/pom.xml

@@ -16,7 +16,7 @@
         <dependency>
             <groupId>io.numaproj.numaflow</groupId>
             <artifactId>numaflow-java</artifactId>
-            <version>0.5.0</version>
+            <version>0.5.1</version>
         </dependency>
 
         <dependency>

examples/src/main/java/io/numaproj/numaflow/examples/sink/simple/SimpleSink.java

@@ -1,25 +1,45 @@
 package io.numaproj.numaflow.examples.sink.simple;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import io.numaproj.numaflow.sinker.Datum;
 import io.numaproj.numaflow.sinker.Response;
+import io.numaproj.numaflow.sinker.ResponseList;
 import io.numaproj.numaflow.sinker.Server;
 import io.numaproj.numaflow.sinker.Sinker;
 import lombok.extern.slf4j.Slf4j;
 
+
 /**
  * This is a simple User Defined Sink example which logs the input message
  */
 
 @Slf4j
 public class SimpleSink extends Sinker {
-
+    private final ObjectMapper mapper = new ObjectMapper();
+    private final ResponseList.ResponseListBuilder responseListBuilder = ResponseList.newBuilder();
     public static void main(String[] args) throws Exception {
         new Server(new SimpleSink()).start();
     }
 
     @Override
-    public Response processMessage(Datum datum) {
-        log.info(new String(datum.getValue()));
-        return Response.responseOK(datum.getId());
+    public void processMessage(Datum datum) {
+        try {
+            String msg = new String(datum.getValue());
+            log.info("Received message: {}", msg);
+            responseListBuilder.addResponse(Response.responseOK(datum.getId()));
+        } catch (Exception e) {
+            responseListBuilder.addResponse(Response.responseFailure(datum.getId(), e.getMessage()));
+        }
+    }
+
+    @Override
+    public ResponseList getResponse() {
+        // Reset the builder after building the response to avoid keeping old responses in memory
+        // this is required as the same sinker instance is used for multiple requests
+        try {
+            return responseListBuilder.build();
+        } finally {
+            responseListBuilder.clearResponses();
+        }
     }
 }

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>io.numaproj.numaflow</groupId>
     <artifactId>numaflow-java</artifactId>
-    <version>0.5.0</version>
+    <version>0.5.1</version>
     <packaging>jar</packaging>
 
     <name>numaflow-java</name>

src/main/java/io/numaproj/numaflow/reducer/Constants.java

@@ -1,24 +1,10 @@
 package io.numaproj.numaflow.reducer;
 
-import io.grpc.Context;
-
 class Constants {
     public static final int DEFAULT_MESSAGE_SIZE = 1024 * 1024 * 64;
 
     public static final String DEFAULT_SOCKET_PATH = "/var/run/numaflow/reduce.sock";
 
-    public static final String WIN_START_KEY = "x-numaflow-win-start-time";
-
-    public static final String WIN_END_KEY = "x-numaflow-win-end-time";
-
-    public static final Context.Key<String> WINDOW_START_TIME = Context.keyWithDefault(
-            WIN_START_KEY,
-                    "");
-
-    public static final Context.Key<String> WINDOW_END_TIME = Context.keyWithDefault(
-            WIN_END_KEY,
-            "");
-
     public static final String EOF = "EOF";
 
     public static final String SUCCESS = "SUCCESS";

src/main/java/io/numaproj/numaflow/reducer/Service.java

@@ -10,6 +10,7 @@
 import io.numaproj.numaflow.reduce.v1.ReduceOuterClass;
 import io.numaproj.numaflow.reducer.metadata.IntervalWindowImpl;
 import io.numaproj.numaflow.reducer.metadata.MetadataImpl;
+import io.numaproj.numaflow.shared.GrpcServerUtils;
 import lombok.extern.slf4j.Slf4j;
 
 import java.time.Instant;
@@ -41,8 +42,8 @@ public StreamObserver<ReduceOuterClass.ReduceRequest> reduceFn(final StreamObser
         }
 
         // get window start and end time from gPRC metadata
-        String winSt = io.numaproj.numaflow.reducer.Constants.WINDOW_START_TIME.get();
-        String winEt = io.numaproj.numaflow.reducer.Constants.WINDOW_END_TIME.get();
+        String winSt = GrpcServerUtils.WINDOW_START_TIME.get();
+        String winEt = GrpcServerUtils.WINDOW_END_TIME.get();
 
         // convert the start and end time to Instant
         Instant startTime = Instant.ofEpochMilli(Long.parseLong(winSt));

src/main/java/io/numaproj/numaflow/shared/GrpcServerUtils.java

@@ -47,7 +47,7 @@ public class GrpcServerUtils {
             WIN_END_KEY,
             "");
 
-    public static final Metadata.Key<String> DATUM_METADATA_WIN_START = io.grpc.Metadata.Key.of(
+    public static final Metadata.Key<String> DATUM_METADATA_WIN_START = Metadata.Key.of(
             WIN_START_KEY,
             Metadata.ASCII_STRING_MARSHALLER);
 
@@ -97,7 +97,7 @@ public static void writeServerInfo(ServerInfoAccessor serverInfoAccessor, String
     }
 
     public static ServerBuilder<?> createServerBuilder(String socketPath, int maxMessageSize) {
-        ServerInterceptor interceptor = new ServerInterceptor() {
+        ServerInterceptor   interceptor = new ServerInterceptor() {
             @Override
             public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
                     ServerCall<ReqT, RespT> call,

src/main/java/io/numaproj/numaflow/sinker/SinkActor.java

@@ -14,7 +14,6 @@
 class SinkActor extends AbstractActor {
 
     private final Sinker sinker;
-    private final ResponseList.ResponseListBuilder responseListBuilder = ResponseList.newBuilder();
 
     public static Props props(Sinker sinker) {
         return Props.create(SinkActor.class, sinker);
@@ -33,12 +32,15 @@ public Receive createReceive() {
                 .build();
     }
 
+    // invokeHandler is called by the sink server when a new message is received.
     private void invokeHandler(HandlerDatum handlerDatum) {
-        responseListBuilder.addResponse(this.sinker.processMessage(handlerDatum));
+        this.sinker.processMessage(handlerDatum);
     }
 
+    // getResult is called by the sink server when EOF is received.
     private void getResult(String eof) {
-        getContext().getParent().tell(responseListBuilder.build(), ActorRef.noSender());
+        ResponseList responseList = this.sinker.getResponse();
+        getContext().getParent().tell(responseList, ActorRef.noSender());
     }
 }
 

src/main/java/io/numaproj/numaflow/sinker/Sinker.java

@@ -9,8 +9,20 @@
 public abstract class Sinker {
     /**
      * method will be used for processing messages.
+     * response for the message should be added to the
+     * response list which will be returned by getResponse
      * @param datum current message to be processed
-     * @return Response to indicate success or failure.
      */
-    public abstract Response processMessage(Datum datum);
+    public abstract void processMessage(Datum datum);
+
+    /**
+     * method will be used for returning the responses.
+     * each message should have a response, if there are
+     * n messages then there should be n responses and
+     * each response should contain the id of the message
+     * Response.responseOK() and Response.responseFailure() can
+     * be used for creating the responses
+     * @return ResponseList which contains the responses
+     */
+    public abstract ResponseList getResponse();
 }

src/test/java/io/numaproj/numaflow/reducer/ServerErrTest.java

@@ -15,13 +15,14 @@
 import io.grpc.testing.GrpcCleanupRule;
 import io.numaproj.numaflow.reduce.v1.ReduceGrpc;
 import io.numaproj.numaflow.reduce.v1.ReduceOuterClass;
+import io.numaproj.numaflow.shared.GrpcServerUtils;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 
-import static io.numaproj.numaflow.reducer.Constants.WIN_END_KEY;
-import static io.numaproj.numaflow.reducer.Constants.WIN_START_KEY;
+import static io.numaproj.numaflow.shared.GrpcServerUtils.WIN_END_KEY;
+import static io.numaproj.numaflow.shared.GrpcServerUtils.WIN_START_KEY;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
@@ -53,9 +54,9 @@ public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
                     ServerCallHandler<ReqT, RespT> next) {
                 final var context =
                         Context.current().withValues(
-                                Constants.WINDOW_START_TIME,
+                                GrpcServerUtils.WINDOW_START_TIME,
                                 headers.get(DATUM_METADATA_WIN_START),
-                                Constants.WINDOW_END_TIME,
+                                GrpcServerUtils.WINDOW_END_TIME,
                                 headers.get(DATUM_METADATA_WIN_END));
                 return Contexts.interceptCall(context, call, headers, next);
             }