This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch branch-0.7
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.7 by this push:
new ad413e59b9 [CELEBORN-2394] Add REST v1 API to unregister shuffles
ad413e59b9 is described below
commit ad413e59b9fbfb66a5719d1ad0b36cf2bca9ac02
Author: Kalvin2077 <[email protected]>
AuthorDate: Mon Aug 3 14:10:18 2026 +0800
[CELEBORN-2394] Add REST v1 API to unregister shuffles
### What changes were proposed in this pull request?
- Add an OpenAPI-based `POST /api/v1/shuffles/unregister` endpoint that
unregisters multiple shuffles for an application in one request.
- Generate Java client support for `ShuffleApi.unregisterShuffles` and
`UnregisterShufflesRequest` from the updated OpenAPI specification.
- Use the existing batch unregister RPC when handling the REST request.
- Add `--unregister-shuffles` support to Celeborn Shell, reusing the
existing `--apps` and `--shuffleIds` options.
- Fix the legacy application resource test to deserialize
`ApplicationsHeartbeatResponse` instead of `ApplicationsResponse`.
### Why are the changes needed?
Shuffle unregistration was previously available only through Celeborn's
internal RPC path. REST API, generated Java client, and Celeborn Shell users
had no corresponding administrative operation.
Supporting multiple shuffle IDs in one request also avoids requiring
callers to issue one request per shuffle and aligns the public interfaces with
Celeborn's existing batch unregister RPC.
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [x] Yes
Users can unregister multiple shuffles through the REST v1 API, the
generated OpenAPI Java client, or Celeborn Shell.
REST request example:
```http
POST /api/v1/shuffles/unregister
Content-Type: application/json
{
"appId": "app1",
"shuffleIds": [1, 2, 3]
}
```
Celeborn Shell example:
```shell
celeborn-cli master --unregister-shuffles --apps app1 --shuffleIds 1,2,3
```
### How was this patch tested?
- Added two tests in `ApiV1MasterResourceSuite` covering batch
unregistration, preservation of unrelated shuffles, idempotency, and invalid
requests.
- Extended `ApiV1OpenapiClientSuite` to cover batch unregistration through
the generated Java client.
- Added Celeborn Shell integration coverage for successful batch
unregistration, missing options, multiple application IDs, malformed shuffle
IDs, and negative shuffle IDs.
- Verified the generated Java sources with:
```shell
mvn -pl openapi/openapi-client -Pgenerate generate-resources -DskipTests
```
- Ran the affected test suites with Maven. The Master REST suite passed 9
tests, the OpenAPI client suite passed 11 tests, and the CLI suite passed 43
tests with one pre-existing HA test canceled.
- Verified formatting with:
```shell
mvn spotless:check -pl cli -am -DskipTests
```
Closes #3770 from Kalvin2077/feat/unregister-shuffle.
Authored-by: Kalvin2077 <[email protected]>
Signed-off-by: Nicholas Jiang <[email protected]>
---
.../apache/celeborn/cli/master/MasterOptions.scala | 5 +
.../celeborn/cli/master/MasterSubcommand.scala | 4 +-
.../celeborn/cli/master/MasterSubcommandImpl.scala | 35 +++--
...stShuffleOptions.scala => ShuffleOptions.scala} | 8 +-
.../celeborn/cli/TestCelebornCliCommands.scala | 26 ++++
docs/celeborn_cli.md | 12 +-
.../celeborn/service/deploy/master/Master.scala | 16 +++
.../master/http/api/v1/ShuffleResource.scala | 34 ++++-
.../http/api/v1/ApiV1MasterResourceSuite.scala | 91 ++++++++++++-
.../apache/celeborn/rest/v1/master/ShuffleApi.java | 78 ++++++++++-
.../rest/v1/model/UnregisterShufflesRequest.java | 150 +++++++++++++++++++++
.../src/main/openapi3/master_rest_v1.yaml | 41 ++++++
.../http/api/v1/ApiV1OpenapiClientSuite.scala | 32 ++++-
13 files changed, 506 insertions(+), 26 deletions(-)
diff --git
a/cli/src/main/scala/org/apache/celeborn/cli/master/MasterOptions.scala
b/cli/src/main/scala/org/apache/celeborn/cli/master/MasterOptions.scala
index 83f6c1d09b..e51f82308f 100644
--- a/cli/src/main/scala/org/apache/celeborn/cli/master/MasterOptions.scala
+++ b/cli/src/main/scala/org/apache/celeborn/cli/master/MasterOptions.scala
@@ -37,6 +37,11 @@ final class MasterOptions {
@Option(names = Array("--show-cluster-shuffles"), description = Array("Show
cluster shuffles"))
private[master] var showClusterShuffles: Boolean = _
+ @Option(
+ names = Array("--unregister-shuffles"),
+ description = Array("Unregister shuffles from the service"))
+ private[master] var unregisterShuffles: Boolean = _
+
@Option(names = Array("--exclude-worker"), description = Array("Exclude
workers by ID"))
private[master] var excludeWorkers: Boolean = _
diff --git
a/cli/src/main/scala/org/apache/celeborn/cli/master/MasterSubcommand.scala
b/cli/src/main/scala/org/apache/celeborn/cli/master/MasterSubcommand.scala
index 3b6d609f80..620490a412 100644
--- a/cli/src/main/scala/org/apache/celeborn/cli/master/MasterSubcommand.scala
+++ b/cli/src/main/scala/org/apache/celeborn/cli/master/MasterSubcommand.scala
@@ -38,7 +38,7 @@ trait MasterSubcommand extends BaseCommand {
private[master] var masterOptions: MasterOptions = _
@ArgGroup(exclusive = false)
- private[master] var reviseLostShuffleOptions: ReviseLostShuffleOptions = _
+ private[master] var shuffleOptions: ShuffleOptions = _
@Mixin
private[master] var commonOptions: CommonOptions = _
@@ -78,6 +78,8 @@ trait MasterSubcommand extends BaseCommand {
private[master] def runShowClusterShuffles: ShufflesResponse
+ private[master] def runUnregisterShuffles: HandleResponse
+
private[master] def runExcludeWorkers: HandleResponse
private[master] def runRemoveExcludedWorkers: HandleResponse
diff --git
a/cli/src/main/scala/org/apache/celeborn/cli/master/MasterSubcommandImpl.scala
b/cli/src/main/scala/org/apache/celeborn/cli/master/MasterSubcommandImpl.scala
index acc735d617..c0e1415c11 100644
---
a/cli/src/main/scala/org/apache/celeborn/cli/master/MasterSubcommandImpl.scala
+++
b/cli/src/main/scala/org/apache/celeborn/cli/master/MasterSubcommandImpl.scala
@@ -35,6 +35,7 @@ class MasterSubcommandImpl extends MasterSubcommand {
if (masterOptions.showClusterApps) log(runShowClusterApps)
if (masterOptions.showClusterAppsInfo) log(runShowClusterAppsInfo)
if (masterOptions.showClusterShuffles) log(runShowClusterShuffles)
+ if (masterOptions.unregisterShuffles) log(runUnregisterShuffles)
if (masterOptions.excludeWorkers) log(runExcludeWorkers)
if (masterOptions.removeExcludedWorkers) log(runRemoveExcludedWorkers)
if (masterOptions.removeWorkersUnavailableInfo)
log(runRemoveWorkersUnavailableInfo)
@@ -79,6 +80,20 @@ class MasterSubcommandImpl extends MasterSubcommand {
private[master] def runShowClusterShuffles: ShufflesResponse =
shuffleApi.getShuffles(commonOptions.getAuthHeader)
+ private[master] def runUnregisterShuffles: HandleResponse = {
+ val (appId, shuffleIds) = getSingleAppShuffleIds
+ if (shuffleIds.asScala.exists(_ < 0)) {
+ throw new ParameterException(
+ spec.commandLine(),
+ "Shuffle ids must be nonnegative.")
+ }
+
+ val request = new UnregisterShufflesRequest()
+ .appId(appId)
+ .shuffleIds(shuffleIds)
+ shuffleApi.unregisterShuffles(request, commonOptions.getAuthHeader)
+ }
+
private[master] def runExcludeWorkers: HandleResponse = {
val workerIds = getWorkerIds
val excludeWorkerRequest = new ExcludeWorkerRequest().add(workerIds)
@@ -260,24 +275,26 @@ class MasterSubcommandImpl extends MasterSubcommand {
private[master] def runShowContainerInfo: ContainerInfo =
defaultApi.getContainerInfo(commonOptions.getAuthHeader)
- override private[master] def reviseLostShuffles: HandleResponse = {
- if (StringUtils.isAnyBlank(commonOptions.apps,
reviseLostShuffleOptions.shuffleIds)) {
+ private def getSingleAppShuffleIds: (String, util.List[Integer]) = {
+ val appId = commonOptions.apps
+ val shuffleIds = Option(shuffleOptions).map(_.shuffleIds).orNull
+ if (StringUtils.isBlank(appId) || shuffleIds == null ||
shuffleIds.isEmpty) {
throw new ParameterException(
spec.commandLine(),
- "Application id and Shuffle ids must be provided for this command.")
+ "Application id and shuffle ids must be provided for this command.")
}
-
- val app = commonOptions.apps
- if (app.contains(",")) {
+ if (appId.contains(",")) {
throw new ParameterException(
spec.commandLine(),
"Only one application id can be provided for this command.")
}
+ (appId, shuffleIds)
+ }
- val shuffleIds = util.Arrays.asList[Integer](
- reviseLostShuffleOptions.shuffleIds.split(",").map(Integer.valueOf): _*)
+ override private[master] def reviseLostShuffles: HandleResponse = {
+ val (appId, shuffleIds) = getSingleAppShuffleIds
val request =
- new ReviseLostShufflesRequest().appId(app).shuffleIds(shuffleIds)
+ new ReviseLostShufflesRequest().appId(appId).shuffleIds(shuffleIds)
applicationApi.reviseLostShuffles(request, commonOptions.getAuthHeader)
}
diff --git
a/cli/src/main/scala/org/apache/celeborn/cli/master/ReviseLostShuffleOptions.scala
b/cli/src/main/scala/org/apache/celeborn/cli/master/ShuffleOptions.scala
similarity index 86%
rename from
cli/src/main/scala/org/apache/celeborn/cli/master/ReviseLostShuffleOptions.scala
rename to cli/src/main/scala/org/apache/celeborn/cli/master/ShuffleOptions.scala
index a0c4963fea..b5767ce444 100644
---
a/cli/src/main/scala/org/apache/celeborn/cli/master/ReviseLostShuffleOptions.scala
+++ b/cli/src/main/scala/org/apache/celeborn/cli/master/ShuffleOptions.scala
@@ -17,13 +17,17 @@
package org.apache.celeborn.cli.master
+import java.util
+
import picocli.CommandLine.Option
-final class ReviseLostShuffleOptions {
+final class ShuffleOptions {
@Option(
names = Array("--shuffleIds"),
+ paramLabel = "shuffleId",
+ split = ",",
description = Array("The shuffle ids to manipulate."))
- private[master] var shuffleIds: String = _
+ private[master] var shuffleIds: util.List[Integer] = _
}
diff --git
a/cli/src/test/scala/org/apache/celeborn/cli/TestCelebornCliCommands.scala
b/cli/src/test/scala/org/apache/celeborn/cli/TestCelebornCliCommands.scala
index 36246e952c..40841b032b 100644
--- a/cli/src/test/scala/org/apache/celeborn/cli/TestCelebornCliCommands.scala
+++ b/cli/src/test/scala/org/apache/celeborn/cli/TestCelebornCliCommands.scala
@@ -219,6 +219,32 @@ class TestCelebornCliCommands extends CelebornFunSuite
with MiniClusterFeature {
captureOutputAndValidateResponse(args, "ShufflesResponse")
}
+ test("master --unregister-shuffles") {
+ val args = prepareMasterArgs() ++ Array(
+ "--unregister-shuffles",
+ "--apps",
+ "app1",
+ "--shuffleIds",
+ "1,2")
+ captureOutputAndValidateResponse(args, "Unregistered shuffles app1-1,
app1-2.")
+ }
+
+ test("master --unregister-shuffles validates inputs") {
+ Seq(
+ Array("--unregister-shuffles", "--shuffleIds", "1,2") ->
+ "Application id and shuffle ids must be provided",
+ Array("--unregister-shuffles", "--apps", "app1") ->
+ "Application id and shuffle ids must be provided",
+ Array("--unregister-shuffles", "--apps", "app1,app2", "--shuffleIds",
"1,2") ->
+ "Only one application id can be provided",
+ Array("--unregister-shuffles", "--apps", "app1", "--shuffleIds",
"1,invalid") ->
+ "Invalid value for option '--shuffleIds'",
+ Array("--unregister-shuffles", "--apps", "app1", "--shuffleIds", "1,-1")
->
+ "Shuffle ids must be nonnegative").foreach { case (command,
expectedError) =>
+ captureErrorAndValidateResponse(prepareMasterArgs() ++ command,
expectedError)
+ }
+ }
+
test("master --show-worker-event-info") {
val args = prepareMasterArgs() :+ "--show-worker-event-info"
captureOutputAndValidateResponse(args, "WorkerEventsResponse")
diff --git a/docs/celeborn_cli.md b/docs/celeborn_cli.md
index 996139aa38..31b893f3a7 100644
--- a/docs/celeborn_cli.md
+++ b/docs/celeborn_cli.md
@@ -87,9 +87,10 @@ Usage: celeborn-cli master [-hV] [--apps=appId]
[--auth-header=authHeader]
v1,k2:v2,k3:v3...] [--worker-ids=w1,w2,w3...]
(--show-masters-info | --show-cluster-apps |
--show-cluster-apps-info | --show-cluster-shuffles |
- --exclude-worker | --remove-excluded-worker |
- --send-worker-event=IMMEDIATELY | DECOMMISSION |
- DECOMMISSION_THEN_IDLE | GRACEFUL | RECOMMISSION |
+ --unregister-shuffles | --exclude-worker |
+ --remove-excluded-worker |
+ --send-worker-event=IMMEDIATELY | DECOMMISSION |
+ DECOMMISSION_THEN_IDLE | GRACEFUL | RECOMMISSION |
NONE | --show-worker-event-info |
--show-lost-workers | --show-excluded-workers |
--show-manual-excluded-workers |
@@ -106,7 +107,7 @@ Usage: celeborn-cli master [-hV] [--apps=appId]
[--auth-header=authHeader]
--revise-lost-shuffles | --delete-apps |
--update-interruption-notices=workerId1=timestamp,
workerId2=timestamp,workerId3=timestamp)
- [[--shuffleIds=<shuffleIds>]]
+ [[--shuffleIds=shuffleId[,shuffleId...]]...]
--add-cluster-alias=alias
Add alias to use in the cli for the given set of
masters
@@ -178,8 +179,9 @@ Usage: celeborn-cli master [-hV] [--apps=appId]
[--auth-header=authHeader]
--show-workers Show registered workers
--show-workers-topology
Show registered workers topology
- --shuffleIds=<shuffleIds>
+ --shuffleIds=shuffleId[,shuffleId...]
The shuffle ids to manipulate.
+ --unregister-shuffles Unregister shuffles from the service
--update-interruption-notices=workerId1=timestamp,workerId2=timestamp,
workerId3=timestamp
Update interruption notices of workers.
diff --git
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
index 4203041b74..75ceb951a9 100644
---
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
+++
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
@@ -1461,6 +1461,22 @@ private[celeborn] class Master(
sb.toString()
}
+ def unregisterShuffles(
+ applicationId: String,
+ shuffleIds: util.List[Integer]): HandleResponse = {
+ val shuffleKeys =
+ shuffleIds.asScala.map(Utils.makeShuffleKey(applicationId,
_)).mkString(", ")
+ val response = self.askSync[PbBatchUnregisterShuffleResponse](
+ BatchUnregisterShuffles(applicationId, shuffleIds,
MasterClient.genRequestId()))
+ val status = StatusCode.fromValue(response.getStatus)
+ val success = status == StatusCode.SUCCESS
+ if (success) {
+ (success, s"Unregistered shuffles $shuffleKeys.")
+ } else {
+ (success, s"Failed to unregister shuffles $shuffleKeys: $status.")
+ }
+ }
+
override def exclude(
addWorkers: Seq[WorkerInfo],
removeWorkers: Seq[WorkerInfo]): HandleResponse = {
diff --git
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ShuffleResource.scala
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ShuffleResource.scala
index ba6ca4cf8c..37b511eba7 100644
---
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ShuffleResource.scala
+++
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ShuffleResource.scala
@@ -18,7 +18,7 @@
package org.apache.celeborn.service.deploy.master.http.api.v1
import java.util
-import javax.ws.rs.{Consumes, GET, Produces}
+import javax.ws.rs.{BadRequestException, Consumes, GET, Path, POST, Produces}
import javax.ws.rs.core.MediaType
import scala.collection.JavaConverters._
@@ -28,15 +28,17 @@ import io.swagger.v3.oas.annotations.media.{Content, Schema}
import io.swagger.v3.oas.annotations.responses.ApiResponse
import io.swagger.v3.oas.annotations.tags.Tag
-import org.apache.celeborn.rest.v1.model.ShufflesResponse
+import org.apache.celeborn.rest.v1.model.{HandleResponse, ShufflesResponse,
UnregisterShufflesRequest}
import org.apache.celeborn.server.common.http.api.ApiRequestContext
import org.apache.celeborn.service.deploy.master.Master
+import
org.apache.celeborn.service.deploy.master.http.api.MasterHttpResourceUtils.ensureMasterIsLeader
@Tag(name = "Shuffle")
@Produces(Array(MediaType.APPLICATION_JSON))
@Consumes(Array(MediaType.APPLICATION_JSON))
class ShuffleResource extends ApiRequestContext {
- private def statusSystem = httpService.asInstanceOf[Master].statusSystem
+ private def master = httpService.asInstanceOf[Master]
+ private def statusSystem = master.statusSystem
@Operation(description =
"List all running shuffle keys of the service. It will return all running
shuffle's key of the cluster.")
@@ -56,4 +58,30 @@ class ShuffleResource extends ApiRequestContext {
}
new ShufflesResponse().shuffleIds(shuffles)
}
+
+ @Operation(description = "Unregister shuffles from the service.")
+ @ApiResponse(
+ responseCode = "200",
+ content = Array(new Content(
+ mediaType = MediaType.APPLICATION_JSON,
+ schema = new Schema(implementation = classOf[HandleResponse]))))
+ @POST
+ @Path("/unregister")
+ def unregisterShuffles(request: UnregisterShufflesRequest): HandleResponse =
+ ensureMasterIsLeader(master) {
+ if (request == null) {
+ throw new BadRequestException("The unregister shuffles request is
required.")
+ }
+ val appId = normalizeParam(request.getAppId)
+ val shuffleIds = request.getShuffleIds
+ if (appId.isEmpty ||
+ shuffleIds == null ||
+ shuffleIds.isEmpty ||
+ shuffleIds.asScala.exists(shuffleId => shuffleId == null || shuffleId
< 0)) {
+ throw new BadRequestException(
+ s"appId(${request.getAppId}) is required and shuffleIds($shuffleIds)
must be a nonempty list of nonnegative ids.")
+ }
+ val (success, message) = master.unregisterShuffles(appId, shuffleIds)
+ new HandleResponse().success(success).message(message)
+ }
}
diff --git
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApiV1MasterResourceSuite.scala
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApiV1MasterResourceSuite.scala
index fec1fce568..0fe21a4b0c 100644
---
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApiV1MasterResourceSuite.scala
+++
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/http/api/v1/ApiV1MasterResourceSuite.scala
@@ -17,12 +17,13 @@
package org.apache.celeborn.service.deploy.master.http.api.v1
-import java.util.Collections
+import java.util.{Arrays, Collections}
import javax.servlet.http.HttpServletResponse
import javax.ws.rs.client.Entity
import javax.ws.rs.core.MediaType
-import org.apache.celeborn.rest.v1.model.{ApplicationsResponse,
ExcludeWorkerRequest, HandleResponse, HostnamesResponse,
RemoveWorkersUnavailableInfoRequest, SendWorkerEventRequest, ShufflesResponse,
TopologyResponse, WorkerEventsResponse, WorkerId, WorkersResponse}
+import org.apache.celeborn.common.util.Utils
+import org.apache.celeborn.rest.v1.model.{ApplicationsHeartbeatResponse,
ExcludeWorkerRequest, HandleResponse, HostnamesResponse,
RemoveWorkersUnavailableInfoRequest, SendWorkerEventRequest, ShufflesResponse,
TopologyResponse, UnregisterShufflesRequest, WorkerEventsResponse, WorkerId,
WorkersResponse}
import org.apache.celeborn.server.common.HttpService
import org.apache.celeborn.server.common.http.api.v1.ApiV1BaseResourceSuite
import org.apache.celeborn.service.deploy.master.{Master, MasterClusterFeature}
@@ -48,10 +49,94 @@ class ApiV1MasterResourceSuite extends
ApiV1BaseResourceSuite with MasterCluster
assert(response.readEntity(classOf[ShufflesResponse]).getShuffleIds.isEmpty)
}
+ test("unregister shuffles preserves other shuffles and is idempotent") {
+ val appId = "unregister-shuffles-app"
+ val shuffleIds = Arrays.asList[Integer](0, 1)
+ val remainingShuffleId = 2
+ shuffleIds.forEach { shuffleId =>
+ master.statusSystem.updateRequestSlotsMeta(
+ Utils.makeShuffleKey(appId, shuffleId),
+ null,
+ Collections.emptyMap[String, java.util.Map[String, Integer]]())
+ }
+ master.statusSystem.updateRequestSlotsMeta(
+ Utils.makeShuffleKey(appId, remainingShuffleId),
+ null,
+ Collections.emptyMap[String, java.util.Map[String, Integer]]())
+ try {
+ shuffleIds.forEach { shuffleId =>
+
assert(master.statusSystem.registeredAppAndShuffles.get(appId).contains(shuffleId))
+ }
+
assert(master.statusSystem.registeredAppAndShuffles.get(appId).contains(remainingShuffleId))
+
+ val request = new
UnregisterShufflesRequest().appId(appId).shuffleIds(shuffleIds)
+ var response =
+
webTarget.path("shuffles/unregister").request(MediaType.APPLICATION_JSON).post(
+ Entity.entity(request, MediaType.APPLICATION_JSON))
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ val handleResponse = response.readEntity(classOf[HandleResponse])
+ assert(handleResponse.getSuccess)
+ shuffleIds.forEach { shuffleId =>
+ assert(handleResponse.getMessage.contains(Utils.makeShuffleKey(appId,
shuffleId)))
+ }
+ val registeredShuffles =
master.statusSystem.registeredAppAndShuffles.get(appId)
+ shuffleIds.forEach { shuffleId =>
+ assert(!registeredShuffles.contains(shuffleId))
+ }
+ assert(registeredShuffles.contains(remainingShuffleId))
+
+ response =
webTarget.path("shuffles/unregister").request(MediaType.APPLICATION_JSON).post(
+ Entity.entity(request, MediaType.APPLICATION_JSON))
+ assert(HttpServletResponse.SC_OK == response.getStatus)
+ assert(response.readEntity(classOf[HandleResponse]).getSuccess)
+
assert(master.statusSystem.registeredAppAndShuffles.get(appId).contains(remainingShuffleId))
+ } finally {
+ master.statusSystem.registeredAppAndShuffles.remove(appId)
+ master.statusSystem.appHeartbeatTime.remove(appId)
+ }
+ }
+
+ test("unregister shuffles validates request") {
+ var response =
+
webTarget.path("shuffles/unregister").request(MediaType.APPLICATION_JSON).post(
+ Entity.entity("null", MediaType.APPLICATION_JSON))
+ assert(HttpServletResponse.SC_BAD_REQUEST == response.getStatus)
+ assert(response.readEntity(classOf[String]).contains("request is
required"))
+
+ response =
+
webTarget.path("shuffles/unregister").request(MediaType.APPLICATION_JSON).post(
+ Entity.entity(
+ new
UnregisterShufflesRequest().shuffleIds(Arrays.asList[Integer](1)),
+ MediaType.APPLICATION_JSON))
+ assert(HttpServletResponse.SC_BAD_REQUEST == response.getStatus)
+ assert(response.readEntity(classOf[String]).contains("appId"))
+
+ response =
webTarget.path("shuffles/unregister").request(MediaType.APPLICATION_JSON).post(
+ Entity.entity(
+ new UnregisterShufflesRequest().appId("
").shuffleIds(Arrays.asList[Integer](1)),
+ MediaType.APPLICATION_JSON))
+ assert(HttpServletResponse.SC_BAD_REQUEST == response.getStatus)
+ assert(response.readEntity(classOf[String]).contains("appId"))
+
+ response =
webTarget.path("shuffles/unregister").request(MediaType.APPLICATION_JSON).post(
+ Entity.entity(
+ new UnregisterShufflesRequest().appId("app"),
+ MediaType.APPLICATION_JSON))
+ assert(HttpServletResponse.SC_BAD_REQUEST == response.getStatus)
+ assert(response.readEntity(classOf[String]).contains("nonempty"))
+
+ response =
webTarget.path("shuffles/unregister").request(MediaType.APPLICATION_JSON).post(
+ Entity.entity(
+ new
UnregisterShufflesRequest().appId("app").shuffleIds(Arrays.asList[Integer](-1)),
+ MediaType.APPLICATION_JSON))
+ assert(HttpServletResponse.SC_BAD_REQUEST == response.getStatus)
+ assert(response.readEntity(classOf[String]).contains("nonnegative"))
+ }
+
test("application resource") {
var response =
webTarget.path("applications").request(MediaType.APPLICATION_JSON).get()
assert(HttpServletResponse.SC_OK == response.getStatus)
-
assert(response.readEntity(classOf[ApplicationsResponse]).getApplications.isEmpty)
+
assert(response.readEntity(classOf[ApplicationsHeartbeatResponse]).getApplications.isEmpty)
response =
webTarget.path("applications/hostnames").request(MediaType.APPLICATION_JSON).get()
assert(HttpServletResponse.SC_OK == response.getStatus)
diff --git
a/openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/master/ShuffleApi.java
b/openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/master/ShuffleApi.java
index 0faaa7c759..030d7b02e4 100644
---
a/openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/master/ShuffleApi.java
+++
b/openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/master/ShuffleApi.java
@@ -25,7 +25,9 @@ import org.apache.celeborn.rest.v1.master.invoker.BaseApi;
import org.apache.celeborn.rest.v1.master.invoker.Configuration;
import org.apache.celeborn.rest.v1.master.invoker.Pair;
+import org.apache.celeborn.rest.v1.model.HandleResponse;
import org.apache.celeborn.rest.v1.model.ShufflesResponse;
+import org.apache.celeborn.rest.v1.model.UnregisterShufflesRequest;
import java.util.ArrayList;
@@ -113,6 +115,80 @@ public class ShuffleApi extends BaseApi {
);
}
+ /**
+ *
+ * Unregister shuffles from the service.
+ * @param unregisterShufflesRequest (required)
+ * @return HandleResponse
+ * @throws ApiException if fails to make API call
+ */
+ public HandleResponse unregisterShuffles(UnregisterShufflesRequest
unregisterShufflesRequest) throws ApiException {
+ return this.unregisterShuffles(unregisterShufflesRequest,
Collections.emptyMap());
+ }
+
+
+ /**
+ *
+ * Unregister shuffles from the service.
+ * @param unregisterShufflesRequest (required)
+ * @param additionalHeaders additionalHeaders for this call
+ * @return HandleResponse
+ * @throws ApiException if fails to make API call
+ */
+ public HandleResponse unregisterShuffles(UnregisterShufflesRequest
unregisterShufflesRequest, Map<String, String> additionalHeaders) throws
ApiException {
+ Object localVarPostBody = unregisterShufflesRequest;
+
+ // verify the required parameter 'unregisterShufflesRequest' is set
+ if (unregisterShufflesRequest == null) {
+ throw new ApiException(400, "Missing the required parameter
'unregisterShufflesRequest' when calling unregisterShuffles");
+ }
+
+ // create path and map variables
+ String localVarPath = "/api/v1/shuffles/unregister";
+
+ StringJoiner localVarQueryStringJoiner = new StringJoiner("&");
+ String localVarQueryParameterBaseName;
+ List<Pair> localVarQueryParams = new ArrayList<Pair>();
+ List<Pair> localVarCollectionQueryParams = new ArrayList<Pair>();
+ Map<String, String> localVarHeaderParams = new HashMap<String, String>();
+ Map<String, String> localVarCookieParams = new HashMap<String, String>();
+ Map<String, Object> localVarFormParams = new HashMap<String, Object>();
+
+
+ localVarHeaderParams.putAll(additionalHeaders);
+
+
+
+ final String[] localVarAccepts = {
+ "application/json"
+ };
+ final String localVarAccept =
apiClient.selectHeaderAccept(localVarAccepts);
+
+ final String[] localVarContentTypes = {
+ "application/json"
+ };
+ final String localVarContentType =
apiClient.selectHeaderContentType(localVarContentTypes);
+
+ String[] localVarAuthNames = new String[] { "basic" };
+
+ TypeReference<HandleResponse> localVarReturnType = new
TypeReference<HandleResponse>() {};
+ return apiClient.invokeAPI(
+ localVarPath,
+ "POST",
+ localVarQueryParams,
+ localVarCollectionQueryParams,
+ localVarQueryStringJoiner.toString(),
+ localVarPostBody,
+ localVarHeaderParams,
+ localVarCookieParams,
+ localVarFormParams,
+ localVarAccept,
+ localVarContentType,
+ localVarAuthNames,
+ localVarReturnType
+ );
+ }
+
@Override
public <T> T invokeAPI(String url, String method, Object request,
TypeReference<T> returnType, Map<String, String> additionalHeaders) throws
ApiException {
String localVarPath = url.replace(apiClient.getBaseURL(), "");
@@ -131,7 +207,7 @@ public class ShuffleApi extends BaseApi {
final String localVarAccept =
apiClient.selectHeaderAccept(localVarAccepts);
final String[] localVarContentTypes = {
-
+ "application/json"
};
final String localVarContentType =
apiClient.selectHeaderContentType(localVarContentTypes);
diff --git
a/openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/model/UnregisterShufflesRequest.java
b/openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/model/UnregisterShufflesRequest.java
new file mode 100644
index 0000000000..29691c0f74
--- /dev/null
+++
b/openapi/openapi-client/src/main/java/org/apache/celeborn/rest/v1/model/UnregisterShufflesRequest.java
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.celeborn.rest.v1.model;
+
+import java.util.Objects;
+import java.util.Arrays;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.fasterxml.jackson.annotation.JsonValue;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+
+/**
+ * UnregisterShufflesRequest
+ */
+@JsonPropertyOrder({
+ UnregisterShufflesRequest.JSON_PROPERTY_APP_ID,
+ UnregisterShufflesRequest.JSON_PROPERTY_SHUFFLE_IDS
+})
[email protected](value =
"org.openapitools.codegen.languages.JavaClientCodegen", comments = "Generator
version: 7.8.0")
+public class UnregisterShufflesRequest {
+ public static final String JSON_PROPERTY_APP_ID = "appId";
+ private String appId;
+
+ public static final String JSON_PROPERTY_SHUFFLE_IDS = "shuffleIds";
+ private List<Integer> shuffleIds = new ArrayList<>();
+
+ public UnregisterShufflesRequest() {
+ }
+
+ public UnregisterShufflesRequest appId(String appId) {
+
+ this.appId = appId;
+ return this;
+ }
+
+ /**
+ * The application id.
+ * @return appId
+ */
+ @javax.annotation.Nonnull
+ @JsonProperty(JSON_PROPERTY_APP_ID)
+ @JsonInclude(value = JsonInclude.Include.ALWAYS)
+
+ public String getAppId() {
+ return appId;
+ }
+
+
+ @JsonProperty(JSON_PROPERTY_APP_ID)
+ @JsonInclude(value = JsonInclude.Include.ALWAYS)
+ public void setAppId(String appId) {
+ this.appId = appId;
+ }
+
+ public UnregisterShufflesRequest shuffleIds(List<Integer> shuffleIds) {
+
+ this.shuffleIds = shuffleIds;
+ return this;
+ }
+
+ public UnregisterShufflesRequest addShuffleIdsItem(Integer shuffleIdsItem) {
+ if (this.shuffleIds == null) {
+ this.shuffleIds = new ArrayList<>();
+ }
+ this.shuffleIds.add(shuffleIdsItem);
+ return this;
+ }
+
+ /**
+ * The shuffle ids.
+ * @return shuffleIds
+ */
+ @javax.annotation.Nonnull
+ @JsonProperty(JSON_PROPERTY_SHUFFLE_IDS)
+ @JsonInclude(value = JsonInclude.Include.ALWAYS)
+
+ public List<Integer> getShuffleIds() {
+ return shuffleIds;
+ }
+
+
+ @JsonProperty(JSON_PROPERTY_SHUFFLE_IDS)
+ @JsonInclude(value = JsonInclude.Include.ALWAYS)
+ public void setShuffleIds(List<Integer> shuffleIds) {
+ this.shuffleIds = shuffleIds;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ UnregisterShufflesRequest unregisterShufflesRequest =
(UnregisterShufflesRequest) o;
+ return Objects.equals(this.appId, unregisterShufflesRequest.appId) &&
+ Objects.equals(this.shuffleIds, unregisterShufflesRequest.shuffleIds);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(appId, shuffleIds);
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("class UnregisterShufflesRequest {\n");
+ sb.append(" appId: ").append(toIndentedString(appId)).append("\n");
+ sb.append(" shuffleIds:
").append(toIndentedString(shuffleIds)).append("\n");
+ sb.append("}");
+ return sb.toString();
+ }
+
+ /**
+ * Convert the given object to string with each line indented by 4 spaces
+ * (except the first line).
+ */
+ private String toIndentedString(Object o) {
+ if (o == null) {
+ return "null";
+ }
+ return o.toString().replace("\n", "\n ");
+ }
+
+}
+
diff --git a/openapi/openapi-client/src/main/openapi3/master_rest_v1.yaml
b/openapi/openapi-client/src/main/openapi3/master_rest_v1.yaml
index 18e7de891d..c76b5f0cfc 100644
--- a/openapi/openapi-client/src/main/openapi3/master_rest_v1.yaml
+++ b/openapi/openapi-client/src/main/openapi3/master_rest_v1.yaml
@@ -171,6 +171,28 @@ paths:
schema:
$ref: '#/components/schemas/ShufflesResponse'
+ /api/v1/shuffles/unregister:
+ post:
+ tags:
+ - Shuffle
+ operationId: unregisterShuffles
+ description: Unregister shuffles from the service.
+ requestBody:
+ required: true
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/UnregisterShufflesRequest'
+ responses:
+ "200":
+ description: The request was handled.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HandleResponse'
+ "400":
+ description: The request is invalid or the target master is not the
leader.
+
/api/v1/masters:
get:
tags:
@@ -782,6 +804,25 @@ components:
items:
type: string
+ UnregisterShufflesRequest:
+ type: object
+ properties:
+ appId:
+ type: string
+ minLength: 1
+ description: The application id.
+ shuffleIds:
+ type: array
+ minItems: 1
+ description: The shuffle ids.
+ items:
+ type: integer
+ format: int32
+ minimum: 0
+ required:
+ - appId
+ - shuffleIds
+
ApplicationHeartbeatData:
type: object
properties:
diff --git
a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1OpenapiClientSuite.scala
b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1OpenapiClientSuite.scala
index 0c1e31543f..ca4adb120b 100644
---
a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1OpenapiClientSuite.scala
+++
b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/http/api/v1/ApiV1OpenapiClientSuite.scala
@@ -17,12 +17,13 @@
package org.apache.celeborn.service.deploy.worker.http.api.v1
-import java.util.Collections
+import java.util.{Arrays, Collections}
import javax.servlet.http.HttpServletResponse
+import org.apache.celeborn.common.util.Utils
import org.apache.celeborn.rest.v1.master._
import org.apache.celeborn.rest.v1.master.invoker._
-import org.apache.celeborn.rest.v1.model.{ExcludeWorkerRequest,
RemoveWorkersUnavailableInfoRequest, SendWorkerEventRequest, WorkerId}
+import org.apache.celeborn.rest.v1.model.{ExcludeWorkerRequest,
RemoveWorkersUnavailableInfoRequest, SendWorkerEventRequest,
UnregisterShufflesRequest, WorkerId}
import org.apache.celeborn.rest.v1.model.SendWorkerEventRequest.EventTypeEnum
class ApiV1OpenapiClientSuite extends ApiV1WorkerOpenapiClientSuite {
@@ -64,6 +65,33 @@ class ApiV1OpenapiClientSuite extends
ApiV1WorkerOpenapiClientSuite {
test("master: shuffle api") {
val api = new ShuffleApi(masterApiClient)
assert(api.getShuffles.getShuffleIds.isEmpty)
+
+ val appId = "openapi-client-unregister-shuffles-app"
+ val shuffleIds = Arrays.asList[Integer](0, 1)
+ shuffleIds.forEach { shuffleId =>
+ master.statusSystem.updateRequestSlotsMeta(
+ Utils.makeShuffleKey(appId, shuffleId),
+ null,
+ Collections.emptyMap[String, java.util.Map[String, Integer]]())
+ }
+ try {
+ shuffleIds.forEach { shuffleId =>
+
assert(api.getShuffles.getShuffleIds.contains(Utils.makeShuffleKey(appId,
shuffleId)))
+ }
+
+ val response =
+ api.unregisterShuffles(
+ new UnregisterShufflesRequest().appId(appId).shuffleIds(shuffleIds))
+ assert(response.getSuccess)
+ shuffleIds.forEach { shuffleId =>
+ val shuffleKey = Utils.makeShuffleKey(appId, shuffleId)
+ assert(response.getMessage.contains(shuffleKey))
+ assert(!api.getShuffles.getShuffleIds.contains(shuffleKey))
+ }
+ } finally {
+ master.statusSystem.registeredAppAndShuffles.remove(appId)
+ master.statusSystem.appHeartbeatTime.remove(appId)
+ }
}
test("master: worker api") {
