-
Notifications
You must be signed in to change notification settings - Fork 590
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
8295: Expose Broker API to step down if not leader r=oleschoenburg a=oleschoenburg ## Description This introduces a new Admin API for the Broker. The only supported request so far is `STEP_DOWN_IF_NOT_LEADER`. The handler has access to the Raft partitions by getting the partition manager injected after partitions are started. The handler is a `PartitionListener` so it subscribes and unsubscribes to messages when it transitions between leader, follower and inactive. This is optional behavior and could be removed as the `stepDownIfNotLeader` is a no-op when the broker is not the leader for this partition. ## Related issues closes #8224 Co-authored-by: Ole Schönburg <ole.schoenburg@gmail.com>
- Loading branch information
Showing
21 changed files
with
539 additions
and
19 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
61 changes: 61 additions & 0 deletions
61
broker/src/main/java/io/camunda/zeebe/broker/bootstrap/AdminApiServiceStep.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| /* | ||
| * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under | ||
| * one or more contributor license agreements. See the NOTICE file distributed | ||
| * with this work for additional information regarding copyright ownership. | ||
| * Licensed under the Zeebe Community License 1.1. You may not use this file | ||
| * except in compliance with the Zeebe Community License 1.1. | ||
| */ | ||
| package io.camunda.zeebe.broker.bootstrap; | ||
|
|
||
| import io.camunda.zeebe.broker.transport.adminapi.AdminApiRequestHandler; | ||
| import io.camunda.zeebe.util.sched.ConcurrencyControl; | ||
| import io.camunda.zeebe.util.sched.future.ActorFuture; | ||
|
|
||
| public class AdminApiServiceStep extends AbstractBrokerStartupStep { | ||
|
|
||
| @Override | ||
| public String getName() { | ||
| return "Admin API"; | ||
| } | ||
|
|
||
| @Override | ||
| void startupInternal( | ||
| final BrokerStartupContext brokerStartupContext, | ||
| final ConcurrencyControl concurrencyControl, | ||
| final ActorFuture<BrokerStartupContext> startupFuture) { | ||
| final var schedulingService = brokerStartupContext.getActorSchedulingService(); | ||
| final var transport = brokerStartupContext.getCommandApiServerTransport(); | ||
| final var handler = new AdminApiRequestHandler(transport); | ||
|
|
||
| concurrencyControl.runOnCompletion( | ||
| schedulingService.submitActor(handler), | ||
| proceed( | ||
| () -> { | ||
| if (brokerStartupContext.getAdminApiService() == null) { | ||
| brokerStartupContext.setAdminApiService(handler); | ||
| } | ||
| startupFuture.complete(brokerStartupContext); | ||
| }, | ||
| startupFuture)); | ||
| } | ||
|
|
||
| @Override | ||
| void shutdownInternal( | ||
| final BrokerStartupContext brokerShutdownContext, | ||
| final ConcurrencyControl concurrencyControl, | ||
| final ActorFuture<BrokerStartupContext> shutdownFuture) { | ||
| final var service = brokerShutdownContext.getAdminApiService(); | ||
| if (service == null) { | ||
| shutdownFuture.complete(brokerShutdownContext); | ||
| return; | ||
| } | ||
| concurrencyControl.runOnCompletion( | ||
| service.closeAsync(), | ||
| proceed( | ||
| () -> { | ||
| brokerShutdownContext.setAdminApiService(null); | ||
| shutdownFuture.complete(brokerShutdownContext); | ||
| }, | ||
| shutdownFuture)); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
68 changes: 68 additions & 0 deletions
68
broker/src/main/java/io/camunda/zeebe/broker/transport/adminapi/AdminApiRequestHandler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| /* | ||
| * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under | ||
| * one or more contributor license agreements. See the NOTICE file distributed | ||
| * with this work for additional information regarding copyright ownership. | ||
| * Licensed under the Zeebe Community License 1.1. You may not use this file | ||
| * except in compliance with the Zeebe Community License 1.1. | ||
| */ | ||
| package io.camunda.zeebe.broker.transport.adminapi; | ||
|
|
||
| import io.atomix.primitive.partition.PartitionId; | ||
| import io.atomix.raft.partition.RaftPartitionGroup; | ||
| import io.camunda.zeebe.broker.partitioning.PartitionManagerImpl; | ||
| import io.camunda.zeebe.broker.transport.ApiRequestHandler; | ||
| import io.camunda.zeebe.broker.transport.ErrorResponseWriter; | ||
| import io.camunda.zeebe.protocol.record.AdminRequestType; | ||
| import io.camunda.zeebe.transport.RequestType; | ||
| import io.camunda.zeebe.transport.impl.AtomixServerTransport; | ||
| import io.camunda.zeebe.util.Either; | ||
|
|
||
| public class AdminApiRequestHandler extends ApiRequestHandler<ApiRequestReader, ApiResponseWriter> { | ||
| private RaftPartitionGroup partitionGroup; | ||
| private final AtomixServerTransport transport; | ||
|
|
||
| public AdminApiRequestHandler(final AtomixServerTransport transport) { | ||
| super(new ApiRequestReader(), new ApiResponseWriter()); | ||
| this.transport = transport; | ||
| } | ||
|
|
||
| @Override | ||
| protected Either<ErrorResponseWriter, ApiResponseWriter> handle( | ||
| final int partitionId, | ||
| final long requestId, | ||
| final ApiRequestReader requestReader, | ||
| final ApiResponseWriter responseWriter, | ||
| final ErrorResponseWriter errorWriter) { | ||
| if (requestReader.getMessageDecoder().type() == AdminRequestType.STEP_DOWN_IF_NOT_PRIMARY) { | ||
| return stepDownIfNotPrimary(responseWriter, partitionId, errorWriter); | ||
| } | ||
| return unknownRequest(errorWriter, requestReader.getMessageDecoder().type()); | ||
| } | ||
|
|
||
| private Either<ErrorResponseWriter, ApiResponseWriter> unknownRequest( | ||
| final ErrorResponseWriter errorWriter, final AdminRequestType type) { | ||
| errorWriter.unsupportedMessage(type, AdminRequestType.values()); | ||
| return Either.left(errorWriter); | ||
| } | ||
|
|
||
| private Either<ErrorResponseWriter, ApiResponseWriter> stepDownIfNotPrimary( | ||
| final ApiResponseWriter responseWriter, | ||
| final int partitionId, | ||
| final ErrorResponseWriter errorWriter) { | ||
| if (partitionGroup == null) { | ||
| errorWriter.partitionLeaderMismatch(partitionId); | ||
| return Either.left(errorWriter); | ||
| } | ||
| final var partition = partitionGroup.getPartition(partitionId); | ||
| partition.stepDownIfNotPrimary(); | ||
|
|
||
| return Either.right(responseWriter); | ||
| } | ||
|
|
||
| public void injectPartitionManager(final PartitionManagerImpl partitionManager) { | ||
| partitionGroup = (RaftPartitionGroup) partitionManager.getPartitionGroup(); | ||
| partitionGroup.getPartitionIds().stream() | ||
| .map(PartitionId::id) | ||
| .forEach(partitionId -> transport.subscribe(partitionId, RequestType.ADMIN, this)); | ||
| } | ||
| } |
31 changes: 31 additions & 0 deletions
31
broker/src/main/java/io/camunda/zeebe/broker/transport/adminapi/ApiRequestReader.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| /* | ||
| * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under | ||
| * one or more contributor license agreements. See the NOTICE file distributed | ||
| * with this work for additional information regarding copyright ownership. | ||
| * Licensed under the Zeebe Community License 1.1. You may not use this file | ||
| * except in compliance with the Zeebe Community License 1.1. | ||
| */ | ||
| package io.camunda.zeebe.broker.transport.adminapi; | ||
|
|
||
| import io.camunda.zeebe.broker.transport.ApiRequestHandler.RequestReader; | ||
| import io.camunda.zeebe.protocol.record.AdminRequestDecoder; | ||
| import io.camunda.zeebe.protocol.record.MessageHeaderDecoder; | ||
| import org.agrona.DirectBuffer; | ||
|
|
||
| public class ApiRequestReader implements RequestReader<AdminRequestDecoder> { | ||
| private final MessageHeaderDecoder headerDecoder = new MessageHeaderDecoder(); | ||
| private final AdminRequestDecoder messageDecoder = new AdminRequestDecoder(); | ||
|
|
||
| @Override | ||
| public void reset() {} | ||
|
|
||
| @Override | ||
| public AdminRequestDecoder getMessageDecoder() { | ||
| return messageDecoder; | ||
| } | ||
|
|
||
| @Override | ||
| public void wrap(final DirectBuffer buffer, final int offset, final int length) { | ||
| messageDecoder.wrapAndApplyHeader(buffer, offset, headerDecoder); | ||
| } | ||
| } |
56 changes: 56 additions & 0 deletions
56
broker/src/main/java/io/camunda/zeebe/broker/transport/adminapi/ApiResponseWriter.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| /* | ||
| * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under | ||
| * one or more contributor license agreements. See the NOTICE file distributed | ||
| * with this work for additional information regarding copyright ownership. | ||
| * Licensed under the Zeebe Community License 1.1. You may not use this file | ||
| * except in compliance with the Zeebe Community License 1.1. | ||
| */ | ||
| package io.camunda.zeebe.broker.transport.adminapi; | ||
|
|
||
| import io.camunda.zeebe.broker.transport.ApiRequestHandler.ResponseWriter; | ||
| import io.camunda.zeebe.protocol.record.AdminResponseEncoder; | ||
| import io.camunda.zeebe.protocol.record.ExecuteQueryResponseEncoder; | ||
| import io.camunda.zeebe.protocol.record.MessageHeaderEncoder; | ||
| import io.camunda.zeebe.transport.ServerOutput; | ||
| import io.camunda.zeebe.transport.impl.ServerResponseImpl; | ||
| import org.agrona.MutableDirectBuffer; | ||
|
|
||
| public class ApiResponseWriter implements ResponseWriter { | ||
| private final MessageHeaderEncoder headerEncoder = new MessageHeaderEncoder(); | ||
| private final AdminResponseEncoder responseEncoder = new AdminResponseEncoder(); | ||
| private final ServerResponseImpl response = new ServerResponseImpl(); | ||
|
|
||
| @Override | ||
| public void tryWriteResponse( | ||
| final ServerOutput output, final int partitionId, final long requestId) { | ||
| try { | ||
| response.reset().writer(this).setPartitionId(partitionId).setRequestId(requestId); | ||
| output.sendResponse(response); | ||
| } finally { | ||
| reset(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void reset() {} | ||
|
|
||
| @Override | ||
| public int getLength() { | ||
| return MessageHeaderEncoder.ENCODED_LENGTH + ExecuteQueryResponseEncoder.BLOCK_LENGTH; | ||
| } | ||
|
|
||
| @Override | ||
| public void write(final MutableDirectBuffer buffer, int offset) { | ||
| headerEncoder.wrap(buffer, offset); | ||
|
|
||
| headerEncoder | ||
| .blockLength(responseEncoder.sbeBlockLength()) | ||
| .templateId(responseEncoder.sbeTemplateId()) | ||
| .schemaId(responseEncoder.sbeSchemaId()) | ||
| .version(responseEncoder.sbeSchemaVersion()); | ||
|
|
||
| offset += headerEncoder.encodedLength(); | ||
|
|
||
| responseEncoder.wrap(buffer, offset); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.