adding initialize routine to launch existing assignments (#86)

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/worker/FunctionMetaDataManager.java

@@ -22,6 +22,7 @@
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.functions.fs.FunctionConfig;
 import org.apache.pulsar.functions.runtime.worker.request.DeregisterRequest;
+import org.apache.pulsar.functions.runtime.worker.request.MarkerRequest;
 import org.apache.pulsar.functions.runtime.worker.request.RequestResult;
 import org.apache.pulsar.functions.runtime.worker.request.ServiceRequest;
 import org.apache.pulsar.functions.runtime.worker.request.ServiceRequestManager;
@@ -32,6 +33,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -55,6 +57,9 @@ public class FunctionMetaDataManager implements AutoCloseable {
 
     private LinkedBlockingQueue<FunctionAction> actionQueue;
 
+    private boolean initializePhase = true;
+    private final String initializeMarkerRequestId = UUID.randomUUID().toString();
+
     public FunctionMetaDataManager(WorkerConfig workerConfig,
                                    ServiceRequestManager serviceRequestManager,
                                    LinkedBlockingQueue<FunctionAction> actionQueue) {
@@ -63,6 +68,19 @@ public FunctionMetaDataManager(WorkerConfig workerConfig,
         this.actionQueue = actionQueue;
     }
 
+    public boolean isInitializePhase() {
+        return initializePhase;
+    }
+
+    public void setInitializePhase(boolean initializePhase) {
+        this.initializePhase = initializePhase;
+    }
+
+    public void sendIntializationMarker() {
+        log.info("sending marking message...");
+        this.serviceRequestManager.submitRequest(new MarkerRequest(this.workerConfig.getWorkerId(), this.initializeMarkerRequestId));
+    }
+
     @Override
     public void close() {
         serviceRequestManager.close();
@@ -208,7 +226,7 @@ public void proccessDeregister(DeregisterRequest deregisterRequest) {
             // check if request is outdated
             if (!isRequestOutdated(deregisterRequest)) {
                 // Check if this worker is suppose to run the function
-                if (isMyRequest(deregisterRequest)) {
+                if (shouldProcessRequest(deregisterRequest)) {
                     // stop running the function
                     insertStopAction(deregisterRequestFs);
                 }
@@ -247,7 +265,7 @@ public void processUpdate(UpdateRequest updateRequest) {
             // Since this is the first time worker has seen function, just put it into internal function metadata store
             addFunctionToFunctionMap(updateRequestFs);
             // Check if this worker is suppose to run the function
-            if (isMyRequest(updateRequest)) {
+            if (shouldProcessRequest(updateRequest)) {
                 insertStartAction(updateRequestFs);
             }
             completeRequest(updateRequest, true);
@@ -265,7 +283,7 @@ public void processUpdate(UpdateRequest updateRequest) {
                 // update the function metadata
                 addFunctionToFunctionMap(updateRequestFs);
                 // check if this worker should run the update
-                if (isMyRequest(updateRequest)) {
+                if (shouldProcessRequest(updateRequest)) {
                     // Update the function
                     insertStartAction(updateRequestFs);
                 }
@@ -299,29 +317,60 @@ private boolean isRequestOutdated(ServiceRequest serviceRequest) {
         return currentFunctionMetaData.getVersion() >= requestFunctionMetaData.getVersion();
     }
 
-    private boolean isMyRequest(ServiceRequest serviceRequest) {
+    private boolean shouldProcessRequest(ServiceRequest serviceRequest) {
         return this.workerConfig.getWorkerId().equals(serviceRequest.getFunctionMetaData().getWorkerId());
     }
 
+    private boolean isSendByMe(ServiceRequest serviceRequest) {
+        return this.workerConfig.getWorkerId().equals(serviceRequest.getWorkerId());
+    }
+
     private void insertStopAction(FunctionMetaData functionMetaData) {
-        FunctionAction functionAction = new FunctionAction();
-        functionAction.setAction(FunctionAction.Action.STOP);
-        functionAction.setFunctionMetaData(functionMetaData);
-        try {
-            actionQueue.put(functionAction);
-        } catch (InterruptedException ex) {
-            throw new RuntimeException("Interrupted while putting action");
+        if (!this.isInitializePhase()) {
+            FunctionAction functionAction = new FunctionAction();
+            functionAction.setAction(FunctionAction.Action.STOP);
+            functionAction.setFunctionMetaData(functionMetaData);
+            try {
+                actionQueue.put(functionAction);
+            } catch (InterruptedException ex) {
+                throw new RuntimeException("Interrupted while putting action");
+            }
         }
     }
 
     private void insertStartAction(FunctionMetaData functionMetaData) {
-        FunctionAction functionAction = new FunctionAction();
-        functionAction.setAction(FunctionAction.Action.START);
-        functionAction.setFunctionMetaData(functionMetaData);
-        try {
-            actionQueue.put(functionAction);
-        } catch (InterruptedException ex) {
-            throw new RuntimeException("Interrupted while putting action");
+        if (!this.isInitializePhase()) {
+            FunctionAction functionAction = new FunctionAction();
+            functionAction.setAction(FunctionAction.Action.START);
+            functionAction.setFunctionMetaData(functionMetaData);
+            try {
+                actionQueue.put(functionAction);
+            } catch (InterruptedException ex) {
+                throw new RuntimeException("Interrupted while putting action");
+            }
+        }
+    }
+
+    private boolean isMyInitializeMarkerRequest(MarkerRequest serviceRequest) {
+        return isSendByMe(serviceRequest) && this.initializeMarkerRequestId.equals(serviceRequest.getRequestId());
+    }
+
+    public void processInitializeMarker(MarkerRequest serviceRequest) {
+        if (isMyInitializeMarkerRequest(serviceRequest)) {
+            this.setInitializePhase(false);
+            log.info("Initializing Metadata state done!");
+            log.info("Launching existing assignments...");
+            // materialize current assignments
+            for (Map<String, Map<String, FunctionMetaData>> i: this.functionMap.values()) {
+                for (Map<String, FunctionMetaData> k : i.values()) {
+                    for (FunctionMetaData functionMetaData : k.values()) {
+                        // if I should run this
+                        if (this.workerConfig.getWorkerId().equals(functionMetaData.getWorkerId())) {
+                            insertStartAction(functionMetaData);
+                        }
+                    }
+                }
+            }
         }
     }
 }

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/worker/FunctionMetaDataTopicTailer.java

@@ -25,7 +25,9 @@
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Reader;
 import org.apache.pulsar.functions.runtime.worker.request.DeregisterRequest;
+import org.apache.pulsar.functions.runtime.worker.request.MarkerRequest;
 import org.apache.pulsar.functions.runtime.worker.request.ServiceRequest;
+import org.apache.pulsar.functions.runtime.worker.request.ServiceRequestManager;
 import org.apache.pulsar.functions.runtime.worker.request.UpdateRequest;
 
 @Slf4j
@@ -43,9 +45,15 @@ public FunctionMetaDataTopicTailer(FunctionMetaDataManager functionMetaDataManag
     }
 
     public void start() {
+        initialize();
         receiveOne();
     }
 
+    public void initialize() {
+        log.info("Initializing Metadata state...");
+        this.functionMetaDataManager.sendIntializationMarker();
+    }
+
     private void receiveOne() {
         reader.readNextAsync()
             .thenAccept(this)
@@ -78,6 +86,9 @@ public void accept(Message msg) {
         }
 
         switch(serviceRequest.getRequestType()) {
+            case MARKER:
+                this.functionMetaDataManager.processInitializeMarker((MarkerRequest) serviceRequest);
+                break;
             case UPDATE:
                 this.functionMetaDataManager.processUpdate((UpdateRequest) serviceRequest);
                 break;

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/worker/request/DeregisterRequest.java

@@ -22,11 +22,10 @@
 
 public class DeregisterRequest extends ServiceRequest{
 
-    private String functionName;
+    private static final long serialVersionUID = 4041129758880596555L;
 
     public DeregisterRequest(String workerId, FunctionMetaData functionMetaData) {
         super(workerId, functionMetaData, ServiceRequestType.DELETE);
-        this.functionName = functionName;
     }
 
     public static DeregisterRequest of(String workerId, FunctionMetaData functionMetaData) {

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/worker/request/MarkerRequest.java

@@ -0,0 +1,16 @@
+package org.apache.pulsar.functions.runtime.worker.request;
+
+public class MarkerRequest extends ServiceRequest{
+    private static final long serialVersionUID = -6074909284607487042L;
+
+    public MarkerRequest(String workerId, String requestId) {
+        super(workerId, null, ServiceRequestType.MARKER, requestId);
+    }
+
+    @Override
+    public String toString() {
+        return "MarkerRequest{"
+                + super.toString()
+                + "}";
+    }
+}

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/worker/request/ServiceRequest.java

@@ -46,7 +46,8 @@ public String toString() {
 
     public enum ServiceRequestType {
         UPDATE,
-        DELETE
+        DELETE,
+        MARKER
     }
 
     public ServiceRequest(String workerId, FunctionMetaData functionMetaData, ServiceRequestType serviceRequestType) {

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/worker/request/UpdateRequest.java

@@ -22,6 +22,8 @@
 
 public class UpdateRequest extends ServiceRequest {
 
+    private static final long serialVersionUID = 4874557910196094342L;
+
     private UpdateRequest(String workerId, FunctionMetaData functionMetaData) {
         super(workerId, functionMetaData, ServiceRequestType.UPDATE);
     }