8000 refactor(inkless:switch): remove the seal and repair operator commands by giuseppelillo · Pull Request #872 · aiven/inkless · GitHub
[go: up one dir, main page]
Skip to content

refactor(inkless:switch): remove the seal and repair operator commands #872

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
giuseppelillo wants to merge 2 commits into
base: main
Choose a base branch
from
+355 −2,682
Draft

refactor(inkless:switch): remove the seal and repair operator commands #872

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b59274e
refactor(inkless:switch): remove the seal and repair operator commands
giuseppelillo Nov 13, 2026
67ab15f
refactor(inkless:switch): replace AlterDisklessSwitch with RearmDiskl…
giuseppelillo Nov 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 12 additions & 53 deletions clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -356,73 +356,32 @@ default DescribeTopicPartitionsResult describeTopicPartitions(Collection<String>
DescribeTopicPartitionsResult describeTopicPartitions(Collection<String> topics, DescribeTopicsOptions options);

/**
* Override the classic-to-diskless switch state of a single partition. This is an operator
* tool that writes the {@code classicToDisklessStartOffset} directly on the controller without
* requiring the caller to be the partition leader.
* Re-arm the pending classic-to-diskless switch of a single partition. This is an operator tool:
* the controller bumps the leader epoch, so the leader seals again and re-runs the switch.
*
* This is a convenience method for {@link #alterDisklessSwitch(String, int, long, AlterDisklessSwitchOptions)}
* with default options. The same rules apply: once the partition's classic prefix is in remote
* storage, the controller rejects every seal offset.
* This is a convenience method for {@link #rearmDisklessSwitch(String, int, RearmDisklessSwitchOptions)}
* with default options.
*
* @param topic The topic name.
* @param partition The partition index.
* @param sealOffset The seal offset to commit: {@code >= 0} forces (re-)sealing at that offset,
* {@code -1} aborts the switch and reverts the partition to classic, and
* {@code -2} re-arms the switch as pending.
* @return The AlterDisklessSwitchResult.
* @return The RearmDisklessSwitchResult.
*/
default AlterDisklessSwitchResult alterDisklessSwitch(String topic, int partition, long sealOffset) {
return alterDisklessSwitch(topic, partition, sealOffset, new AlterDisklessSwitchOptions());
default RearmDisklessSwitchResult rearmDisklessSwitch(String topic, int partition) {
return rearmDisklessSwitch(topic, partition, new RearmDisklessSwitchOptions());
}

/**
* Override the classic-to-diskless switch state of a single partition. This is an operator
* tool that writes the {@code classicToDisklessStartOffset} directly on the controller without
* requiring the caller to be the partition leader.
* Re-arm the pending classic-to-diskless switch of a single partition. This is an operator tool:
* the controller bumps the leader epoch, so the leader seals again and re-runs the switch.
*
* Once the partition's classic prefix is in remote storage ({@code classicPrefixInRemoteStorage} is
* set), the controller rejects every seal offset and the result fails with
* {@code InvalidRequestException}.
* If the switch of the partition isn't pending, the result fails with {@code InvalidRequestException}.
*
* @param topic The topic name.
* @param partition The partition index.
* @param sealOffset The seal offset to commit: {@code >= 0} forces (re-)sealing at that offset,
* {@code -1} aborts the switch and reverts the partition to classic, and
* {@code -2} re-arms the switch as pending.
* @param options The options to use.
* @return The AlterDisklessSwitchResult.
*/
AlterDisklessSwitchResult alterDisklessSwitch(String topic, int partition, long sealOffset, AlterDisklessSwitchOptions options);

/**
* Reconcile the control-plane diskless log entry of a single partition with the seal committed in
* KRaft. This is an operator tool; the request must be sent to the partition leader, which owns the
* control-plane connection, so the caller supplies its broker id.
*
* This is a convenience method for {@link #repairDisklessLog(String, int, int, RepairDisklessLogOptions)}
* with default options.
*
* @param topic The topic name.
* @param partition The partition index.
* @param brokerId The id of the partition leader broker to send the request to.
* @return The RepairDisklessLogResult.
*/
default RepairDisklessLogResult repairDisklessLog(String topic, int partition, int brokerId) {
return repairDisklessLog(topic, partition, brokerId, new RepairDisklessLogOptions());
}

/**
* Reconcile the control-plane diskless log entry of a single partition with the seal committed in
* KRaft. This is an operator tool; the request must be sent to the partition leader, which owns the
* control-plane connection, so the caller supplies its broker id.
*
* @param topic The topic name.
* @param partition The partition index.
* @param brokerId The id of the partition leader broker to send the request to.
* @param options The options to use.
* @return The RepairDisklessLogResult.
* @return The RearmDisklessSwitchResult.
*/
RepairDisklessLogResult repairDisklessLog(String topic, int partition, int brokerId, RepairDisklessLogOptions options);
RearmDisklessSwitchResult rearmDisklessSwitch(String topic, int partition, RearmDisklessSwitchOptions options);

/**
* Get information about the nodes in the cluster, using the default options.
Expand Down
43 changes: 0 additions & 43 deletions clients/src/main/java/org/apache/kafka/clients/admin/AlterDisklessSwitchOptions.java
View file
Open in desktop

This file was deleted.

9 changes: 2 additions & 7 deletions clients/src/main/java/org/apache/kafka/clients/admin/ForwardingAdmin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,13 +84,8 @@ public DescribeTopicPartitionsResult describeTopicPartitions(Collection<String>
}

@Override
public AlterDisklessSwitchResult alterDisklessSwitch(String topic, int partition, long sealOffset, AlterDisklessSwitchOptions options) {
return delegate.alterDisklessSwitch(topic, partition, sealOffset, options);
}

@Override
public RepairDisklessLogResult repairDisklessLog(String topic, int partition, int brokerId, RepairDisklessLogOptions options) {
return delegate.repairDisklessLog(topic, partition, brokerId, options);
public RearmDisklessSwitchResult rearmDisklessSwitch(String topic, int partition, RearmDisklessSwitchOptions options) {
return delegate.rearmDisklessSwitch(topic, partition, options);
}

@Override
Expand Down
63 changes: 10 additions & 53 deletions clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
View file
Open in desktop
10BC8
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.internals.KafkaFutureImpl;
import org.apache.kafka.common.message.AddRaftVoterRequestData;
import org.apache.kafka.common.message.AlterDisklessSwitchRequestData;
import org.apache.kafka.common.message.RearmDisklessSwitchRequestData;
import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData;
import org.apache.kafka.common.message.AlterPartitionReassignmentsRequestData.ReassignableTopic;
import org.apache.kafka.common.message.AlterReplicaLogDirsRequestData;
Expand Down Expand Up @@ -168,7 +168,6 @@
import org.apache.kafka.common.message.MetadataRequestData;
import org.apache.kafka.common.message.RemoveRaftVoterRequestData;
import org.apache.kafka.common.message.RenewDelegationTokenRequestData;
import org.apache.kafka.common.message.RepairDisklessLogRequestData;
import org.apache.kafka.common.message.UnregisterBrokerRequestData;
import org.apache.kafka.common.message.UpdateFeaturesRequestData;
import org.apache.kafka.common.message.UpdateFeaturesResponseData.UpdatableFeatureResult;
Expand All @@ -189,8 +188,8 @@
import org.apache.kafka.common.requests.AddRaftVoterResponse;
import org.apache.kafka.common.requests.AlterClientQuotasRequest;
import org.apache.kafka.common.requests.AlterClientQuotasResponse;
import org.apache.kafka.common.requests.AlterDisklessSwitchRequest;
import org.apache.kafka.common.requests.AlterDisklessSwitchResponse;
import org.apache.kafka.common.requests.RearmDisklessSwitchRequest;
import org.apache.kafka.common.requests.RearmDisklessSwitchResponse;
import org.apache.kafka.common.requests.AlterPartitionReassignmentsRequest;
import org.apache.kafka.common.requests.AlterPartitionReassignmentsResponse;
import org.apache.kafka.common.requests.AlterReplicaLogDirsRequest;
Expand Down Expand Up @@ -251,8 +250,6 @@
import org.apache.kafka.common.requests.RemoveRaftVoterResponse;
import org.apache.kafka.common.requests.RenewDelegationTokenRequest;
import org.apache.kafka.common.requests.RenewDelegationTokenResponse;
import org.apache.kafka.common.requests.RepairDisklessLogRequest;
import org.apache.kafka.common.requests.RepairDisklessLogResponse;
import org.apache.kafka.common.requests.UnregisterBrokerRequest;
import org.apache.kafka.common.requests.UnregisterBrokerResponse;
import org.apache.kafka.common.requests.UpdateFeaturesRequest;
Expand Down Expand Up @@ -4907,63 +4904,23 @@ void handleFailure(Throwable throwable) {
}

@Override
public AlterDisklessSwitchResult alterDisklessSwitch(String topic, int partition, long sealOffset,
AlterDisklessSwitchOptions options) {
public RearmDisklessSwitchResult rearmDisklessSwitch(String topic, int partition, RearmDisklessSwitchOptions options) {
final KafkaFutureImpl<Void> future = new KafkaFutureImpl<>();
final long now = time.milliseconds();
final Call call = new Call("alterDisklessSwitch", calcDeadlineMs(now, options.timeoutMs()),
final Call call = new Call("rearmDisklessSwitch", calcDeadlineMs(now, options.timeoutMs()),
new LeastLoadedBrokerOrActiveKController()) {

@Override
AlterDisklessSwitchRequest.Builder createRequest(int timeoutMs) {
AlterDisklessSwitchRequestData data = new AlterDisklessSwitchRequestData()
.setTopicName(topic)
.setPartitionIndex(partition)
.setSealOffset(sealOffset)
.setClearProducerStates(options.clearProducerStates());
return new AlterDisklessSwitchRequest.Builder(data);
}

@Override
void handleResponse(AbstractResponse abstractResponse) {
final AlterDisklessSwitchResponse response = (AlterDisklessSwitchResponse) abstractResponse;
Errors error = Errors.forCode(response.data().errorCode());
if (error == Errors.NONE) {
future.complete(null);
} else {
future.completeExceptionally(error.exception(response.data().errorMessage()));
}
}

@Override
void handleFailure(Throwable throwable) {
future.completeExceptionally(throwable);
}
};
runnable.call(call, now);
return new AlterDisklessSwitchResult(future);
}

@Override
public RepairDisklessLogResult repairDisklessLog(String topic, int partition, int brokerId,
RepairDisklessLogOptions options) {
final KafkaFutureImpl<Void> future = new KafkaFutureImpl<>();
final long now = time.milliseconds();
// The control-plane write happens on the partition leader, so send directly to that broker.
final Call call = new Call("repairDisklessLog", calcDeadlineMs(now, options.timeoutMs()),
new ConstantNodeIdProvider(brokerId)) {

@Override
RepairDisklessLogRequest.Builder createRequest(int timeoutMs) {
RepairDisklessLogRequestData data = new RepairDisklessLogRequestData()
RearmDisklessSwitchRequest.Builder createRequest(int timeoutMs) {
RearmDisklessSwitchRequestData data = new RearmDisklessSwitchRequestData()
.setTopicName(topic)
.setPartitionIndex(partition);
return new RepairDisklessLogRequest.Builder(data);
return new RearmDisklessSwitchRequest.Builder(data);
}

@Override
void handleResponse(AbstractResponse abstractResponse) {
final RepairDisklessLogResponse response = (RepairDisklessLogResponse) abstractResponse;
final RearmDisklessSwitchResponse response = (RearmDisklessSwitchResponse) abstractResponse;
Errors error = Errors.forCode(response.data().errorCode());
if (error == Errors.NONE) {
future.complete(null);
Expand All @@ -4978,7 +4935,7 @@ void handleFailure(Throwable throwable) {
}
};
runnable.call(call, now);
return new RepairDisklessLogResult(future);
return new RearmDisklessSwitchResult(future);
}

@Override
Expand Down
4 changes: 2 additions & 2 deletions ...ients/admin/RepairDisklessLogOptions.java → ...nts/admin/RearmDisklessSwitchOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,10 @@
import org.apache.kafka.common.annotation.InterfaceStability;

/**
* Options for {@link Admin#repairDisklessLog(String, int, int, RepairDisklessLogOptions)}.
* Options for {@link Admin#rearmDisklessSwitch(String, int, long, RearmDisklessSwitchOptions)}.
*
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Unstable
public class RepairDisklessLogOptions extends AbstractOptions<RepairDisklessLogOptions> {
public class RearmDisklessSwitchOptions extends AbstractOptions<RearmDisklessSwitchOptions> {
}
6 changes: 3 additions & 3 deletions ...ents/admin/AlterDisklessSwitchResult.java → ...ents/admin/RearmDisklessSwitchResult.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,15 +21,15 @@
import org.apache.kafka.common.annotation.InterfaceStability;

/**
* The result of the {@link Admin#alterDisklessSwitch(String, int, long, AlterDisklessSwitchOptions)} call.
* The result of the {@link Admin#rearmDisklessSwitch(String, int, long, RearmDisklessSwitchOptions)} call.
*
* The API of this class is evolving, see {@link Admin} for details.
*/
@InterfaceStability.Unstable
public class AlterDisklessSwitchResult {
public class RearmDisklessSwitchResult {
private final KafkaFuture<Void> future;

AlterDisklessSwitchResult(final KafkaFuture<Void> future) {
RearmDisklessSwitchResult(final KafkaFuture<Void> future) {
this.future = future;
}

Expand Down
42 changes: 0 additions & 42 deletions clients/src/main/java/org/apache/kafka/clients/admin/RepairDisklessLogResult.java
View file
Open in desktop

This file was deleted.

3 changes: 1 addition & 2 deletions clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,8 +139,7 @@ public enum ApiKeys {
ALTER_SHARE_GROUP_OFFSETS(ApiMessageType.ALTER_SHARE_GROUP_OFFSETS),
DELETE_SHARE_GROUP_OFFSETS(ApiMessageType.DELETE_SHARE_GROUP_OFFSETS),
INIT_DISKLESS_LOG(ApiMessageType.INIT_DISKLESS_LOG, true),
ALTER_DISKLESS_SWITCH(ApiMessageType.ALTER_DISKLESS_SWITCH, false, true),
REPAIR_DISKLESS_LOG(ApiMessageType.REPAIR_DISKLESS_LOG, false, false),
REARM_DISKLESS_SWITCH(ApiMessageType.REARM_DISKLESS_SWITCH, false, true),
SET_CLASSIC_PREFIX_IN_REMOTE_STORAGE(ApiMessageType.SET_CLASSIC_PREFIX_IN_REMOTE_STORAGE, true);


Expand Down
6 changes: 2 additions & 4 deletions clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -356,10 +356,8 @@ private static AbstractRequest doParseRequest(ApiKeys apiKey, short apiVersion,
return DeleteShareGroupOffsetsRequest.parse(readable, apiVersion);
case INIT_DISKLESS_LOG:
return InitDisklessLogRequest.parse(readable, apiVersion);
case ALTER_DISKLESS_SWITCH:
return AlterDisklessSwitchRequest.parse(readable, apiVersion);
case REPAIR_DISKLESS_LOG:
return RepairDisklessLogRequest.parse(readable, apiVersion);
case REARM_DISKLESS_SWITCH:
return RearmDisklessSwitchRequest.parse(readable, apiVersion);
case SET_CLASSIC_PREFIX_IN_REMOTE_STORAGE:
return SetClassicPrefixInRemoteStorageRequest.parse(readable, apiVersion);
default:
Expand Down
6 changes: 2 additions & 4 deletions clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -293,10 +293,8 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, Readable readable,
return DeleteShareGroupOffsetsResponse.parse(readable, version);
case INIT_DISKLESS_LOG:
return InitDisklessLogResponse.parse(readable, version);
case ALTER_DISKLESS_SWITCH:
return AlterDisklessSwitchResponse.parse(readable, version);
case REPAIR_DISKLESS_LOG:
return RepairDisklessLogResponse.parse(readable, version);
case REARM_DISKLESS_SWITCH:
return RearmDisklessSwitchResponse.parse(readable, version);
case SET_CLASSIC_PREFIX_IN_REMOTE_STORAGE:
return SetClassicPrefixInRemoteStorageResponse.parse(readable, version);
default:
Expand Down
Loading
Loading
0