Improved API for function worker liveliness probe

Author: mukesh-ctds

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

@@ -81,6 +81,8 @@ public class FunctionMetaDataManager implements AutoCloseable {
     @Getter
     private CompletableFuture<Void> isInitialized = new CompletableFuture<>();
 
+    private boolean isFunctionWorkerAlive = true;
+
     public FunctionMetaDataManager(WorkerConfig workerConfig,
                                    SchedulerManager schedulerManager,
                                    PulsarClient pulsarClient,
@@ -243,6 +245,10 @@ public synchronized void updateFunctionOnLeader(FunctionMetaData functionMetaDat
                 needsScheduling = processUpdate(functionMetaData);
             }
         } catch (Exception e) {
+            if (e.getCause() instanceof PulsarClientException.ProducerFencedException) {
+                log.error("Function worker status has been set to false due to ProducerFencedException.");
+                this.isFunctionWorkerAlive = false;
+            }
             log.error("Could not write into Function Metadata topic", e);
             throw new IllegalStateException("Internal Error updating function at the leader", e);
         }
@@ -500,4 +506,8 @@ private void initializeTailer() throws PulsarClientException {
         this.functionMetaDataTopicTailer.start();
         log.info("MetaData Manager Tailer started");
     }
+
+    public boolean checkLiveliness() {
+        return this.isFunctionWorkerAlive;
+    }
 }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java

@@ -1885,4 +1885,10 @@ protected ValidatableFunctionPackage getBuiltinFunctionPackage(String archive) {
         }
         return null;
     }
+
+    @Override
+    public boolean checkLiveliness() {
+        FunctionMetaDataManager functionMetaDataManager = worker().getFunctionMetaDataManager();
+        return functionMetaDataManager.checkLiveliness();
+    }
 }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java

@@ -69,8 +69,6 @@
 @Slf4j
 public class FunctionsImpl extends ComponentImpl implements Functions<PulsarWorkerService> {
 
-    private boolean isFunctionWorkerAlive = true;
-
     public FunctionsImpl(Supplier<PulsarWorkerService> workerServiceSupplier) {
         super(workerServiceSupplier, Function.FunctionDetails.ComponentType.FUNCTION);
     }
@@ -695,8 +693,6 @@ public void updateFunctionOnWorkerLeader(final String tenant,
         try {
             functionMetaDataManager.updateFunctionOnLeader(functionMetaData, delete);
         } catch (IllegalStateException e) {
-            log.error("Function worker status has been set to false due to ProducerFencedException.");
-            this.isFunctionWorkerAlive = false;
             throw new RestException(Response.Status.INTERNAL_SERVER_ERROR, e.getMessage());
         } catch (IllegalArgumentException e) {
             throw new RestException(Response.Status.BAD_REQUEST, e.getMessage());
@@ -792,8 +788,4 @@ private Function.FunctionDetails validateUpdateRequestParams(final String tenant
             }
         }
     }
-
-    public boolean checkLiveliness() {
-        return this.isFunctionWorkerAlive;
-    }
 }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java

@@ -445,7 +445,8 @@ public Response healthCheck() {
                     .entity("There is IllegalStateException, Service is not running. Need to restart.")
                     .build();
         } else {
-            return Response.ok("There is no IllegalStateException, Service is running.")
+            return Response.status(Response.Status.OK)
+                    .entity("There is no IllegalStateException, Service is running.")
                     .build();
         }
     }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Component.java

@@ -90,4 +90,6 @@ StreamingOutput downloadFunction(String tenant, String namespace, String compone
     List<ConnectorDefinition> getListOfConnectors();
 
     void reloadConnectors(AuthenticationParameters authParams);
+
+    boolean checkLiveliness();
 }

pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/service/api/Functions.java

@@ -102,6 +102,4 @@ FunctionInstanceStatusData getFunctionInstanceStatus(String tenant,
     void reloadBuiltinFunctions(AuthenticationParameters authParams) throws IOException;
 
     List<FunctionDefinition> getBuiltinFunctions(AuthenticationParameters authParams);
-
-    boolean checkLiveliness();
 }