[CELEBORN-1589] Ensure master is leader for some POST request APIs

master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala

@@ -1364,7 +1364,7 @@ private[celeborn] class Master(
     }
   }
 
-  private def isMasterActive: Int = {
+  def isMasterActive: Int = {
     // use int rather than bool for better monitoring on dashboard
     val isActive =
       if (conf.haEnabled) {

master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/WorkerResource.scala

@@ -73,10 +73,15 @@ class WorkerResource extends ApiRequestContext {
   @POST
   @Path("/exclude")
   def excludeWorker(request: ExcludeWorkerRequest): HandleResponse = {
-    val (success, msg) = httpService.exclude(
-      request.getAdd.asScala.map(ApiUtils.toWorkerInfo).toSeq,
-      request.getRemove.asScala.map(ApiUtils.toWorkerInfo).toSeq)
-    new HandleResponse().success(success).message(msg)
+    if (master.isMasterActive == 1) {
+      val (success, msg) = httpService.exclude(
+        request.getAdd.asScala.map(ApiUtils.toWorkerInfo).toSeq,
+        request.getRemove.asScala.map(ApiUtils.toWorkerInfo).toSeq)
+      new HandleResponse().success(success).message(msg)
+    } else {
+      throw new BadRequestException(
+        s"This operation can only be done from a master that has the LEADER role.")
+    }
   }
 
   @ApiResponse(
@@ -88,9 +93,15 @@ class WorkerResource extends ApiRequestContext {
   @POST
   @Path("/remove_unavailable")
   def removeWorkersUnavailableInfo(request: RemoveWorkersUnavailableInfoRequest): HandleResponse = {
-    val (success, msg) = master.removeWorkersUnavailableInfo(
-      request.getWorkers.asScala.map(ApiUtils.toWorkerInfo).toSeq)
-    new HandleResponse().success(success).message(msg)
+    if (master.isMasterActive == 1) {
+      val (success, msg) = master.removeWorkersUnavailableInfo(
+        request.getWorkers.asScala.map(ApiUtils.toWorkerInfo).toSeq)
+      new HandleResponse().success(success).message(msg)
+    } else {
+      throw new BadRequestException(
+        s"This operation can only be done from a master that has the LEADER role.")
+    }
+
   }
 
   @ApiResponse(
@@ -124,23 +135,28 @@ class WorkerResource extends ApiRequestContext {
   @POST
   @Path("/events")
   def sendWorkerEvents(request: SendWorkerEventRequest): HandleResponse = {
-    if (request.getEventType == SendWorkerEventRequest.EventTypeEnum.NONE || request.getWorkers.isEmpty) {
-      throw new BadRequestException(
-        s"eventType(${request.getEventType}) and workers(${request.getWorkers}) are required")
-    }
-    val workers = request.getWorkers.asScala.map(ApiUtils.toWorkerInfo).toSeq
-    val (filteredWorkers, unknownWorkers) = workers.partition(statusSystem.workers.contains)
-    if (filteredWorkers.isEmpty) {
+    if (master.isMasterActive == 1) {
+      if (request.getEventType == SendWorkerEventRequest.EventTypeEnum.NONE || request.getWorkers.isEmpty) {
+        throw new BadRequestException(
+          s"eventType(${request.getEventType}) and workers(${request.getWorkers}) are required")
+      }
+      val workers = request.getWorkers.asScala.map(ApiUtils.toWorkerInfo).toSeq
+      val (filteredWorkers, unknownWorkers) = workers.partition(statusSystem.workers.contains)
+      if (filteredWorkers.isEmpty) {
+        throw new BadRequestException(
+          s"None of the workers are known: ${unknownWorkers.map(_.readableAddress).mkString(", ")}")
+      }
+      val (success, msg) = httpService.handleWorkerEvent(request.getEventType.toString, workers)
+      val finalMsg =
+        if (unknownWorkers.isEmpty) {
+          msg
+        } else {
+          s"${msg}\n(Unknown workers: ${unknownWorkers.map(_.readableAddress).mkString(", ")})"
+        }
+      new HandleResponse().success(success).message(finalMsg)
+    } else {
       throw new BadRequestException(
-        s"None of the workers are known: ${unknownWorkers.map(_.readableAddress).mkString(", ")}")
+        s"This operation can only be done from a master that has the LEADER role.")
     }
-    val (success, msg) = httpService.handleWorkerEvent(request.getEventType.toString, workers)
-    val finalMsg =
-      if (unknownWorkers.isEmpty) {
-        msg
-      } else {
-        s"${msg}\n(Unknown workers: ${unknownWorkers.map(_.readableAddress).mkString(", ")})"
-      }
-    new HandleResponse().success(success).message(finalMsg)
   }
 }