Add readiness api for the worker leader (apache#7601)

* Added upgrade notes * Add new end point to verify leader readiness * Address feedback * Address feedback * Add new end point to verify leader readiness * Address feedback * Address feedback Co-authored-by: Sanjeev Kulkarni <[email protected]>
rkaw92 · Jul 24, 2020 · b5aa109 · b5aa109
1 parent 835f506
commit b5aa109
Show file tree

Hide file tree

Showing 3 changed files with 39 additions and 0 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Worker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Worker.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.broker.admin.v2;
 
+import com.sun.org.apache.xpath.internal.operations.Bool;
 import io.swagger.annotations.ApiOperation;
 import io.swagger.annotations.ApiResponse;
 import io.swagger.annotations.ApiResponses;
@@ -129,4 +130,17 @@ public List<ConnectorDefinition> getConnectorsList() throws IOException {
     public void rebalance() {
         worker.rebalance(uri.getRequestUri(), clientAppId());
     }
+
+    @GET
+    @ApiOperation(
+            value = "Checks if this node is the leader and is ready to service requests",
+            response = Boolean.class
+    )
+    @ApiResponses(value = {
+            @ApiResponse(code = 503, message = "Worker service is not running")
+    })
+    @Path("/cluster/leader/ready")
+    public Boolean isLeaderReady() {
+        return worker.isLeaderReady(clientAppId());
+    }
 }
diff --git a/...unctions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java b/...unctions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java
@@ -229,4 +229,16 @@ public void rebalance(final URI uri, final String clientRole) {
             throw new WebApplicationException(Response.temporaryRedirect(redirect).build());
         }
     }
+
+    public Boolean isLeaderReady(final String clientRole) {
+        if (!isWorkerServiceAvailable()) {
+            throwUnavailableException();
+        }
+        if (worker().getLeaderService().isLeader()) {
+            return true;
+        } else {
+            throwUnavailableException();
+            return false; // make compiler happy
+        }
+    }
 }
diff --git a/...ker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java b/...ker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/WorkerApiV2Resource.java
@@ -156,4 +156,17 @@ public List<ConnectorDefinition> getConnectorsList() throws IOException {
     public void rebalance() {
         worker.rebalance(uri.getRequestUri(), clientAppId());
     }
+
+    @GET
+    @ApiOperation(
+            value = "Checks if this node is the leader and is ready to service requests",
+            response = Boolean.class
+    )
+    @ApiResponses(value = {
+            @ApiResponse(code = 503, message = "Worker service is not running")
+    })
+    @Path("/cluster/leader/ready")
+    public Boolean isLeaderReady() {
+        return worker.isLeaderReady(clientAppId());
+    }
 }