You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "joshua2519 (via GitHub)" <gi...@apache.org> on 2026/02/09 12:41:11 UTC
[PR] KAFKA-19914: Move AlterPartitionManager to server module [kafka]
joshua2519 opened a new pull request, #21432:
URL: https://github.com/apache/kafka/pull/21432
Migrate AlterPartitionManager from Scala to Java in the `org.apache.kafka.server.partition` package, including the interface, `DefaultAlterPartitionManager` implementation, and `AlterPartitionItem` record. Also migrate `AlterPartitionManagerTest` and move `MockAlterPartitionManager` from `TestUtils.scala` to `org.apache.kafka.server.util`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19914: Move AlterPartitionManager to server module [kafka]
Posted by "joshua2519 (via GitHub)" <gi...@apache.org>.
joshua2519 commented on code in PR #21432:
URL: https://github.com/apache/kafka/pull/21432#discussion_r3026085376
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.ControllerInformation;
+import org.apache.kafka.server.NodeToControllerChannelManagerImpl;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.config.AbstractKafkaConfig;
+import org.apache.kafka.server.util.Scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class DefaultAlterPartitionManager implements AlterPartitionManager {
+ private static final Logger log = LoggerFactory.getLogger(DefaultAlterPartitionManager.class);
+ private final NodeToControllerChannelManager controllerChannelManager;
+ private final Scheduler scheduler;
+ private final int brokerId;
+ private final Supplier<Long> brokerEpochSupplier;
+
+ // Used to allow only one pending ISR update per partition (visible for testing)
+ final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem> unsentIsrUpdates = new ConcurrentHashMap<>();
+
+ // Used to allow only one in-flight request at a time
+ private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
+
+ private DefaultAlterPartitionManager(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ this.controllerChannelManager = controllerChannelManager;
+ this.scheduler = scheduler;
+ this.brokerId = brokerId;
+ this.brokerEpochSupplier = brokerEpochSupplier;
+ }
+
+ // For testing
+ public static DefaultAlterPartitionManager create(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, Time time, int brokerId, Supplier<Long> brokerEpochSupplier) {
Review Comment:
Good catch! Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19914: Move AlterPartitionManager to server module [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21432:
URL: https://github.com/apache/kafka/pull/21432#issuecomment-3976262677
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19914: Move AlterPartitionManager to server module [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21432:
URL: https://github.com/apache/kafka/pull/21432#issuecomment-4087642009
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19914: Move AlterPartitionManager to server module [kafka]
Posted by "joshua2519 (via GitHub)" <gi...@apache.org>.
joshua2519 commented on code in PR #21432:
URL: https://github.com/apache/kafka/pull/21432#discussion_r3026088066
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.ControllerInformation;
+import org.apache.kafka.server.NodeToControllerChannelManagerImpl;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.config.AbstractKafkaConfig;
+import org.apache.kafka.server.util.Scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class DefaultAlterPartitionManager implements AlterPartitionManager {
+ private static final Logger log = LoggerFactory.getLogger(DefaultAlterPartitionManager.class);
+ private final NodeToControllerChannelManager controllerChannelManager;
+ private final Scheduler scheduler;
+ private final int brokerId;
+ private final Supplier<Long> brokerEpochSupplier;
+
+ // Used to allow only one pending ISR update per partition (visible for testing)
+ final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem> unsentIsrUpdates = new ConcurrentHashMap<>();
+
+ // Used to allow only one in-flight request at a time
+ private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
+
+ private DefaultAlterPartitionManager(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ this.controllerChannelManager = controllerChannelManager;
+ this.scheduler = scheduler;
+ this.brokerId = brokerId;
+ this.brokerEpochSupplier = brokerEpochSupplier;
+ }
+
+ // For testing
+ public static DefaultAlterPartitionManager create(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, Time time, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ return new DefaultAlterPartitionManager(controllerChannelManager, scheduler, brokerId, brokerEpochSupplier);
+ }
+
+ public static DefaultAlterPartitionManager create(AbstractKafkaConfig config,
+ Scheduler scheduler,
+ Supplier<ControllerInformation> controllerNodeProvider,
+ Time time,
+ Metrics metrics,
+ String threadNamePrefix,
+ Supplier<Long> brokerEpochSupplier) {
+ NodeToControllerChannelManager channelManager = new NodeToControllerChannelManagerImpl(
+ controllerNodeProvider,
+ time,
+ metrics,
+ config,
+ "alter-partition",
+ threadNamePrefix,
+ Long.MAX_VALUE
+ );
+
+ return new DefaultAlterPartitionManager(channelManager, scheduler, config.brokerId(), brokerEpochSupplier);
+ }
+
+ @Override
+ public void start() {
+ controllerChannelManager.start();
+ }
+
+ @Override
+ public void shutdown() throws InterruptedException {
+ controllerChannelManager.shutdown();
+ }
+
+ @Override
+ public CompletableFuture<LeaderAndIsr> submit(TopicIdPartition topicIdPartition,
+ LeaderAndIsr leaderAndIsr) {
+ CompletableFuture<LeaderAndIsr> future = new CompletableFuture<>();
+ AlterPartitionItem alterPartitionItem = new AlterPartitionItem(topicIdPartition, leaderAndIsr, future);
+ boolean enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition(), alterPartitionItem) == null;
+ if (enqueued) {
+ maybePropagateIsrChanges();
+ } else {
+ future.completeExceptionally(new OperationNotAttemptedException(String.format(
+ "Failed to enqueue ISR change state %s for partition %s", leaderAndIsr, topicIdPartition)));
+ }
+ return future;
+ }
+
+ void maybePropagateIsrChanges() {
+ // Send all pending items if there is not already a request in-flight.
+ if (!unsentIsrUpdates.isEmpty() && inflightRequest.compareAndSet(false, true)) {
+ // Copy current unsent ISRs but don't remove from the map, they get cleared in the response handler
+ List<AlterPartitionItem> inflightAlterPartitionItems = new ArrayList<>(unsentIsrUpdates.values());
+ sendRequest(inflightAlterPartitionItems);
+ }
+ }
+
+ void clearInFlightRequest() {
+ if (!inflightRequest.compareAndSet(true, false)) {
+ log.warn("Attempting to clear AlterPartition in-flight flag when no apparent request is in-flight");
+ }
+ }
+
+ private void sendRequest(List<AlterPartitionItem> inflightAlterPartitionItems) {
+ long brokerEpoch = brokerEpochSupplier.get();
+ AlterPartitionRequest.Builder request = buildRequest(inflightAlterPartitionItems, brokerEpoch);
+ log.debug("Sending AlterPartition to controller {}", request);
+
+ // We will not time out AlterPartition request, instead letting it retry indefinitely
+ // until a response is received, or a new LeaderAndIsr overwrites the existing isrState
+ // which causes the response for those partitions to be ignored.
+ controllerChannelManager.sendRequest(request,
+ new ControllerRequestCompletionHandler() {
+ @Override
+ public void onComplete(ClientResponse response) {
+ log.debug("Received AlterPartition response {}", response);
+ Errors error;
+ try {
+ if (response.authenticationException() != null) {
+ // For now, we treat authentication errors as retriable. We use the
+ // `NETWORK_EXCEPTION` error code for lack of a good alternative.
+ // Note that `NodeToControllerChannelManager` will still log the
+ // authentication errors so that users have a chance to fix the problem.
+ error = Errors.NETWORK_EXCEPTION;
+ } else if (response.versionMismatch() != null) {
+ error = Errors.UNSUPPORTED_VERSION;
+ } else {
+ error = handleAlterPartitionResponse(
+ (AlterPartitionResponse) response.responseBody(),
+ brokerEpoch,
+ inflightAlterPartitionItems
+ );
+ }
+ } finally {
+ // clear the flag so future requests can proceed
+ clearInFlightRequest();
+ }
+
+ // check if we need to send another request right away
+ if (Objects.requireNonNull(error) == Errors.NONE) {
+ // In the normal case, check for pending updates to send immediately
+ maybePropagateIsrChanges();
+ } else {
+ // If we received a top-level error from the controller, retry the request in the near future
+ scheduler.scheduleOnce("send-alter-partition", () -> maybePropagateIsrChanges(), 50);
+ }
+ }
+
+ @Override
+ public void onTimeout() {
+ throw new IllegalStateException("Encountered unexpected timeout when sending AlterPartition to the controller");
+ }
+ });
+ }
+
+ /**
+ * Builds an AlterPartition request.
+ * <p>
+ * While building the request, we don't know which version of the AlterPartition API is
+ * supported by the controller. The final decision is taken when the AlterPartitionRequest
+ * is built in the network client based on the advertised api versions of the controller.
+ *
+ * @return an AlterPartitionRequest.Builder with the provided parameters.
+ */
+ private AlterPartitionRequest.Builder buildRequest(
+ List<AlterPartitionItem> inflightAlterPartitionItems,
+ long brokerEpoch) {
+ AlterPartitionRequestData message = new AlterPartitionRequestData()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(brokerEpoch);
+
+ inflightAlterPartitionItems.stream()
+ .collect(Collectors.groupingBy(item -> item.topicIdPartition().topicId()))
+ .forEach((topicId, items) -> {
+ AlterPartitionRequestData.TopicData topicData = new AlterPartitionRequestData.TopicData().setTopicId(topicId);
+ message.topics().add(topicData);
+
+ items.forEach(item -> {
+ AlterPartitionRequestData.PartitionData partitionData =
+ new AlterPartitionRequestData.PartitionData()
+ .setPartitionIndex(item.topicIdPartition().partitionId())
+ .setLeaderEpoch(item.leaderAndIsr().leaderEpoch())
+ .setNewIsrWithEpochs(item.leaderAndIsr().isrWithBrokerEpoch())
+ .setPartitionEpoch(item.leaderAndIsr().partitionEpoch());
+
+ partitionData.setLeaderRecoveryState(item.leaderAndIsr().leaderRecoveryState().value());
+
+ topicData.partitions().add(partitionData);
+ });
+
+ });
+
+ return new AlterPartitionRequest.Builder(message);
+ }
+
+ private Errors handleAlterPartitionResponse(AlterPartitionResponse alterPartitionResponse,
+ long sentBrokerEpoch,
+ List<AlterPartitionItem> inflightAlterPartitionItems) {
+ AlterPartitionResponseData data = alterPartitionResponse.data();
+ Errors e = Errors.forCode(data.errorCode());
Review Comment:
Done
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.ControllerInformation;
+import org.apache.kafka.server.NodeToControllerChannelManagerImpl;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.config.AbstractKafkaConfig;
+import org.apache.kafka.server.util.Scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class DefaultAlterPartitionManager implements AlterPartitionManager {
+ private static final Logger log = LoggerFactory.getLogger(DefaultAlterPartitionManager.class);
+ private final NodeToControllerChannelManager controllerChannelManager;
+ private final Scheduler scheduler;
+ private final int brokerId;
+ private final Supplier<Long> brokerEpochSupplier;
+
+ // Used to allow only one pending ISR update per partition (visible for testing)
+ final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem> unsentIsrUpdates = new ConcurrentHashMap<>();
+
+ // Used to allow only one in-flight request at a time
+ private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
+
+ private DefaultAlterPartitionManager(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ this.controllerChannelManager = controllerChannelManager;
+ this.scheduler = scheduler;
+ this.brokerId = brokerId;
+ this.brokerEpochSupplier = brokerEpochSupplier;
+ }
+
+ // For testing
+ public static DefaultAlterPartitionManager create(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, Time time, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ return new DefaultAlterPartitionManager(controllerChannelManager, scheduler, brokerId, brokerEpochSupplier);
+ }
+
+ public static DefaultAlterPartitionManager create(AbstractKafkaConfig config,
+ Scheduler scheduler,
+ Supplier<ControllerInformation> controllerNodeProvider,
+ Time time,
+ Metrics metrics,
+ String threadNamePrefix,
+ Supplier<Long> brokerEpochSupplier) {
+ NodeToControllerChannelManager channelManager = new NodeToControllerChannelManagerImpl(
+ controllerNodeProvider,
+ time,
+ metrics,
+ config,
+ "alter-partition",
+ threadNamePrefix,
+ Long.MAX_VALUE
+ );
+
+ return new DefaultAlterPartitionManager(channelManager, scheduler, config.brokerId(), brokerEpochSupplier);
+ }
+
+ @Override
+ public void start() {
+ controllerChannelManager.start();
+ }
+
+ @Override
+ public void shutdown() throws InterruptedException {
+ controllerChannelManager.shutdown();
+ }
+
+ @Override
+ public CompletableFuture<LeaderAndIsr> submit(TopicIdPartition topicIdPartition,
+ LeaderAndIsr leaderAndIsr) {
+ CompletableFuture<LeaderAndIsr> future = new CompletableFuture<>();
+ AlterPartitionItem alterPartitionItem = new AlterPartitionItem(topicIdPartition, leaderAndIsr, future);
+ boolean enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition(), alterPartitionItem) == null;
+ if (enqueued) {
+ maybePropagateIsrChanges();
+ } else {
+ future.completeExceptionally(new OperationNotAttemptedException(String.format(
+ "Failed to enqueue ISR change state %s for partition %s", leaderAndIsr, topicIdPartition)));
+ }
+ return future;
+ }
+
+ void maybePropagateIsrChanges() {
+ // Send all pending items if there is not already a request in-flight.
+ if (!unsentIsrUpdates.isEmpty() && inflightRequest.compareAndSet(false, true)) {
+ // Copy current unsent ISRs but don't remove from the map, they get cleared in the response handler
+ List<AlterPartitionItem> inflightAlterPartitionItems = new ArrayList<>(unsentIsrUpdates.values());
+ sendRequest(inflightAlterPartitionItems);
+ }
+ }
+
+ void clearInFlightRequest() {
+ if (!inflightRequest.compareAndSet(true, false)) {
+ log.warn("Attempting to clear AlterPartition in-flight flag when no apparent request is in-flight");
+ }
+ }
+
+ private void sendRequest(List<AlterPartitionItem> inflightAlterPartitionItems) {
+ long brokerEpoch = brokerEpochSupplier.get();
+ AlterPartitionRequest.Builder request = buildRequest(inflightAlterPartitionItems, brokerEpoch);
+ log.debug("Sending AlterPartition to controller {}", request);
+
+ // We will not time out AlterPartition request, instead letting it retry indefinitely
+ // until a response is received, or a new LeaderAndIsr overwrites the existing isrState
+ // which causes the response for those partitions to be ignored.
+ controllerChannelManager.sendRequest(request,
+ new ControllerRequestCompletionHandler() {
+ @Override
+ public void onComplete(ClientResponse response) {
+ log.debug("Received AlterPartition response {}", response);
+ Errors error;
+ try {
+ if (response.authenticationException() != null) {
+ // For now, we treat authentication errors as retriable. We use the
+ // `NETWORK_EXCEPTION` error code for lack of a good alternative.
+ // Note that `NodeToControllerChannelManager` will still log the
+ // authentication errors so that users have a chance to fix the problem.
+ error = Errors.NETWORK_EXCEPTION;
+ } else if (response.versionMismatch() != null) {
+ error = Errors.UNSUPPORTED_VERSION;
+ } else {
+ error = handleAlterPartitionResponse(
+ (AlterPartitionResponse) response.responseBody(),
+ brokerEpoch,
+ inflightAlterPartitionItems
+ );
+ }
+ } finally {
+ // clear the flag so future requests can proceed
+ clearInFlightRequest();
+ }
+
+ // check if we need to send another request right away
+ if (Objects.requireNonNull(error) == Errors.NONE) {
+ // In the normal case, check for pending updates to send immediately
+ maybePropagateIsrChanges();
+ } else {
+ // If we received a top-level error from the controller, retry the request in the near future
+ scheduler.scheduleOnce("send-alter-partition", () -> maybePropagateIsrChanges(), 50);
+ }
+ }
+
+ @Override
+ public void onTimeout() {
+ throw new IllegalStateException("Encountered unexpected timeout when sending AlterPartition to the controller");
+ }
+ });
+ }
+
+ /**
+ * Builds an AlterPartition request.
+ * <p>
+ * While building the request, we don't know which version of the AlterPartition API is
+ * supported by the controller. The final decision is taken when the AlterPartitionRequest
+ * is built in the network client based on the advertised api versions of the controller.
+ *
+ * @return an AlterPartitionRequest.Builder with the provided parameters.
+ */
+ private AlterPartitionRequest.Builder buildRequest(
+ List<AlterPartitionItem> inflightAlterPartitionItems,
+ long brokerEpoch) {
+ AlterPartitionRequestData message = new AlterPartitionRequestData()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(brokerEpoch);
+
+ inflightAlterPartitionItems.stream()
+ .collect(Collectors.groupingBy(item -> item.topicIdPartition().topicId()))
+ .forEach((topicId, items) -> {
+ AlterPartitionRequestData.TopicData topicData = new AlterPartitionRequestData.TopicData().setTopicId(topicId);
+ message.topics().add(topicData);
+
+ items.forEach(item -> {
+ AlterPartitionRequestData.PartitionData partitionData =
+ new AlterPartitionRequestData.PartitionData()
+ .setPartitionIndex(item.topicIdPartition().partitionId())
+ .setLeaderEpoch(item.leaderAndIsr().leaderEpoch())
+ .setNewIsrWithEpochs(item.leaderAndIsr().isrWithBrokerEpoch())
+ .setPartitionEpoch(item.leaderAndIsr().partitionEpoch());
+
+ partitionData.setLeaderRecoveryState(item.leaderAndIsr().leaderRecoveryState().value());
+
+ topicData.partitions().add(partitionData);
+ });
+
+ });
+
+ return new AlterPartitionRequest.Builder(message);
+ }
+
+ private Errors handleAlterPartitionResponse(AlterPartitionResponse alterPartitionResponse,
+ long sentBrokerEpoch,
+ List<AlterPartitionItem> inflightAlterPartitionItems) {
+ AlterPartitionResponseData data = alterPartitionResponse.data();
+ Errors e = Errors.forCode(data.errorCode());
+ switch (e) {
+ case STALE_BROKER_EPOCH -> log.warn("Broker had a stale broker epoch ({}), retrying.", sentBrokerEpoch);
+ case CLUSTER_AUTHORIZATION_FAILED -> log.error("Broker is not authorized to send AlterPartition to controller",
+ Errors.CLUSTER_AUTHORIZATION_FAILED.exception("Broker is not authorized to send AlterPartition to controller"));
+ case NONE -> {
+ // Collect partition-level responses to pass to the callbacks
+ Map<TopicIdPartition, LeaderAndIsr> successResponses = new HashMap<>();
+ Map<TopicIdPartition, Errors> errorResponses = new HashMap<>();
+ data.topics().forEach(topic ->
+ topic.partitions().forEach(partition -> {
+ TopicIdPartition tp = new TopicIdPartition(topic.topicId(), partition.partitionIndex());
+ Errors apiError = Errors.forCode(partition.errorCode());
+ log.debug("Controller successfully handled AlterPartition request for {}: {}", tp, partition);
+ if (apiError == Errors.NONE) {
+ Optional<LeaderRecoveryState> leaderRecoveryStateOpt =
+ LeaderRecoveryState.optionalOf(partition.leaderRecoveryState());
+ if (leaderRecoveryStateOpt.isPresent()) {
+ successResponses.put(tp, new LeaderAndIsr(
+ partition.leaderId(),
+ partition.leaderEpoch(),
+ partition.isr(),
+ leaderRecoveryStateOpt.get(),
+ partition.partitionEpoch()
+ ));
+ } else {
+ log.error("Controller returned an invalid leader recovery state ({}) for {}: {}", partition.leaderRecoveryState(), tp, partition);
+ errorResponses.put(tp, Errors.UNKNOWN_SERVER_ERROR);
+ }
+ } else {
+ errorResponses.put(tp, apiError);
+ }
+ }
+ ));
+ // Iterate across the items we sent rather than what we received to ensure we run the callback even if a
+ // partition was somehow erroneously excluded from the response. Note that these callbacks are run from
+ // the leaderIsrUpdateLock write lock in Partition#sendAlterPartitionRequest
+ inflightAlterPartitionItems.forEach(inflightAlterPartition -> {
+ if (successResponses.get(inflightAlterPartition.topicIdPartition()) != null) {
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19914: Move AlterPartitionManager to server module [kafka]
Posted by "joshua2519 (via GitHub)" <gi...@apache.org>.
joshua2519 commented on code in PR #21432:
URL: https://github.com/apache/kafka/pull/21432#discussion_r3026086543
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.ControllerInformation;
+import org.apache.kafka.server.NodeToControllerChannelManagerImpl;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.config.AbstractKafkaConfig;
+import org.apache.kafka.server.util.Scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class DefaultAlterPartitionManager implements AlterPartitionManager {
+ private static final Logger log = LoggerFactory.getLogger(DefaultAlterPartitionManager.class);
+ private final NodeToControllerChannelManager controllerChannelManager;
+ private final Scheduler scheduler;
+ private final int brokerId;
+ private final Supplier<Long> brokerEpochSupplier;
+
+ // Used to allow only one pending ISR update per partition (visible for testing)
+ final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem> unsentIsrUpdates = new ConcurrentHashMap<>();
+
+ // Used to allow only one in-flight request at a time
+ private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
+
+ private DefaultAlterPartitionManager(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ this.controllerChannelManager = controllerChannelManager;
+ this.scheduler = scheduler;
+ this.brokerId = brokerId;
+ this.brokerEpochSupplier = brokerEpochSupplier;
+ }
+
+ // For testing
+ public static DefaultAlterPartitionManager create(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, Time time, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ return new DefaultAlterPartitionManager(controllerChannelManager, scheduler, brokerId, brokerEpochSupplier);
+ }
+
+ public static DefaultAlterPartitionManager create(AbstractKafkaConfig config,
+ Scheduler scheduler,
+ Supplier<ControllerInformation> controllerNodeProvider,
+ Time time,
+ Metrics metrics,
+ String threadNamePrefix,
+ Supplier<Long> brokerEpochSupplier) {
+ NodeToControllerChannelManager channelManager = new NodeToControllerChannelManagerImpl(
+ controllerNodeProvider,
+ time,
+ metrics,
+ config,
+ "alter-partition",
+ threadNamePrefix,
+ Long.MAX_VALUE
+ );
+
+ return new DefaultAlterPartitionManager(channelManager, scheduler, config.brokerId(), brokerEpochSupplier);
+ }
+
+ @Override
+ public void start() {
+ controllerChannelManager.start();
+ }
+
+ @Override
+ public void shutdown() throws InterruptedException {
+ controllerChannelManager.shutdown();
+ }
+
+ @Override
+ public CompletableFuture<LeaderAndIsr> submit(TopicIdPartition topicIdPartition,
+ LeaderAndIsr leaderAndIsr) {
+ CompletableFuture<LeaderAndIsr> future = new CompletableFuture<>();
+ AlterPartitionItem alterPartitionItem = new AlterPartitionItem(topicIdPartition, leaderAndIsr, future);
+ boolean enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition(), alterPartitionItem) == null;
+ if (enqueued) {
+ maybePropagateIsrChanges();
+ } else {
+ future.completeExceptionally(new OperationNotAttemptedException(String.format(
+ "Failed to enqueue ISR change state %s for partition %s", leaderAndIsr, topicIdPartition)));
+ }
+ return future;
+ }
+
+ void maybePropagateIsrChanges() {
+ // Send all pending items if there is not already a request in-flight.
+ if (!unsentIsrUpdates.isEmpty() && inflightRequest.compareAndSet(false, true)) {
+ // Copy current unsent ISRs but don't remove from the map, they get cleared in the response handler
+ List<AlterPartitionItem> inflightAlterPartitionItems = new ArrayList<>(unsentIsrUpdates.values());
+ sendRequest(inflightAlterPartitionItems);
+ }
+ }
+
+ void clearInFlightRequest() {
+ if (!inflightRequest.compareAndSet(true, false)) {
+ log.warn("Attempting to clear AlterPartition in-flight flag when no apparent request is in-flight");
+ }
+ }
+
+ private void sendRequest(List<AlterPartitionItem> inflightAlterPartitionItems) {
+ long brokerEpoch = brokerEpochSupplier.get();
+ AlterPartitionRequest.Builder request = buildRequest(inflightAlterPartitionItems, brokerEpoch);
+ log.debug("Sending AlterPartition to controller {}", request);
+
+ // We will not time out AlterPartition request, instead letting it retry indefinitely
+ // until a response is received, or a new LeaderAndIsr overwrites the existing isrState
+ // which causes the response for those partitions to be ignored.
+ controllerChannelManager.sendRequest(request,
+ new ControllerRequestCompletionHandler() {
+ @Override
+ public void onComplete(ClientResponse response) {
+ log.debug("Received AlterPartition response {}", response);
+ Errors error;
+ try {
+ if (response.authenticationException() != null) {
+ // For now, we treat authentication errors as retriable. We use the
+ // `NETWORK_EXCEPTION` error code for lack of a good alternative.
+ // Note that `NodeToControllerChannelManager` will still log the
+ // authentication errors so that users have a chance to fix the problem.
+ error = Errors.NETWORK_EXCEPTION;
+ } else if (response.versionMismatch() != null) {
+ error = Errors.UNSUPPORTED_VERSION;
+ } else {
+ error = handleAlterPartitionResponse(
+ (AlterPartitionResponse) response.responseBody(),
+ brokerEpoch,
+ inflightAlterPartitionItems
+ );
+ }
+ } finally {
+ // clear the flag so future requests can proceed
+ clearInFlightRequest();
+ }
+
+ // check if we need to send another request right away
+ if (Objects.requireNonNull(error) == Errors.NONE) {
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19914: Move AlterPartitionManager to server module [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison merged PR #21432:
URL: https://github.com/apache/kafka/pull/21432
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19914: Move AlterPartitionManager to server module [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #21432:
URL: https://github.com/apache/kafka/pull/21432#discussion_r2974671019
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.ControllerInformation;
+import org.apache.kafka.server.NodeToControllerChannelManagerImpl;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.config.AbstractKafkaConfig;
+import org.apache.kafka.server.util.Scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class DefaultAlterPartitionManager implements AlterPartitionManager {
+ private static final Logger log = LoggerFactory.getLogger(DefaultAlterPartitionManager.class);
+ private final NodeToControllerChannelManager controllerChannelManager;
+ private final Scheduler scheduler;
+ private final int brokerId;
+ private final Supplier<Long> brokerEpochSupplier;
+
+ // Used to allow only one pending ISR update per partition (visible for testing)
+ final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem> unsentIsrUpdates = new ConcurrentHashMap<>();
+
+ // Used to allow only one in-flight request at a time
+ private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
+
+ private DefaultAlterPartitionManager(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ this.controllerChannelManager = controllerChannelManager;
+ this.scheduler = scheduler;
+ this.brokerId = brokerId;
+ this.brokerEpochSupplier = brokerEpochSupplier;
+ }
+
+ // For testing
+ public static DefaultAlterPartitionManager create(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, Time time, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ return new DefaultAlterPartitionManager(controllerChannelManager, scheduler, brokerId, brokerEpochSupplier);
+ }
+
+ public static DefaultAlterPartitionManager create(AbstractKafkaConfig config,
+ Scheduler scheduler,
+ Supplier<ControllerInformation> controllerNodeProvider,
+ Time time,
+ Metrics metrics,
+ String threadNamePrefix,
+ Supplier<Long> brokerEpochSupplier) {
+ NodeToControllerChannelManager channelManager = new NodeToControllerChannelManagerImpl(
+ controllerNodeProvider,
+ time,
+ metrics,
+ config,
+ "alter-partition",
+ threadNamePrefix,
+ Long.MAX_VALUE
+ );
+
+ return new DefaultAlterPartitionManager(channelManager, scheduler, config.brokerId(), brokerEpochSupplier);
+ }
+
+ @Override
+ public void start() {
+ controllerChannelManager.start();
+ }
+
+ @Override
+ public void shutdown() throws InterruptedException {
+ controllerChannelManager.shutdown();
+ }
+
+ @Override
+ public CompletableFuture<LeaderAndIsr> submit(TopicIdPartition topicIdPartition,
+ LeaderAndIsr leaderAndIsr) {
+ CompletableFuture<LeaderAndIsr> future = new CompletableFuture<>();
+ AlterPartitionItem alterPartitionItem = new AlterPartitionItem(topicIdPartition, leaderAndIsr, future);
+ boolean enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition(), alterPartitionItem) == null;
+ if (enqueued) {
+ maybePropagateIsrChanges();
+ } else {
+ future.completeExceptionally(new OperationNotAttemptedException(String.format(
+ "Failed to enqueue ISR change state %s for partition %s", leaderAndIsr, topicIdPartition)));
+ }
+ return future;
+ }
+
+ void maybePropagateIsrChanges() {
+ // Send all pending items if there is not already a request in-flight.
+ if (!unsentIsrUpdates.isEmpty() && inflightRequest.compareAndSet(false, true)) {
+ // Copy current unsent ISRs but don't remove from the map, they get cleared in the response handler
+ List<AlterPartitionItem> inflightAlterPartitionItems = new ArrayList<>(unsentIsrUpdates.values());
+ sendRequest(inflightAlterPartitionItems);
+ }
+ }
+
+ void clearInFlightRequest() {
+ if (!inflightRequest.compareAndSet(true, false)) {
+ log.warn("Attempting to clear AlterPartition in-flight flag when no apparent request is in-flight");
+ }
+ }
+
+ private void sendRequest(List<AlterPartitionItem> inflightAlterPartitionItems) {
+ long brokerEpoch = brokerEpochSupplier.get();
+ AlterPartitionRequest.Builder request = buildRequest(inflightAlterPartitionItems, brokerEpoch);
+ log.debug("Sending AlterPartition to controller {}", request);
+
+ // We will not time out AlterPartition request, instead letting it retry indefinitely
+ // until a response is received, or a new LeaderAndIsr overwrites the existing isrState
+ // which causes the response for those partitions to be ignored.
+ controllerChannelManager.sendRequest(request,
+ new ControllerRequestCompletionHandler() {
+ @Override
+ public void onComplete(ClientResponse response) {
+ log.debug("Received AlterPartition response {}", response);
+ Errors error;
+ try {
+ if (response.authenticationException() != null) {
+ // For now, we treat authentication errors as retriable. We use the
+ // `NETWORK_EXCEPTION` error code for lack of a good alternative.
+ // Note that `NodeToControllerChannelManager` will still log the
+ // authentication errors so that users have a chance to fix the problem.
+ error = Errors.NETWORK_EXCEPTION;
+ } else if (response.versionMismatch() != null) {
+ error = Errors.UNSUPPORTED_VERSION;
+ } else {
+ error = handleAlterPartitionResponse(
+ (AlterPartitionResponse) response.responseBody(),
+ brokerEpoch,
+ inflightAlterPartitionItems
+ );
+ }
+ } finally {
+ // clear the flag so future requests can proceed
+ clearInFlightRequest();
+ }
+
+ // check if we need to send another request right away
+ if (Objects.requireNonNull(error) == Errors.NONE) {
Review Comment:
Do we need `Objects.requireNonNull()`? It seems `error` can't be `null`
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.ControllerInformation;
+import org.apache.kafka.server.NodeToControllerChannelManagerImpl;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.config.AbstractKafkaConfig;
+import org.apache.kafka.server.util.Scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class DefaultAlterPartitionManager implements AlterPartitionManager {
+ private static final Logger log = LoggerFactory.getLogger(DefaultAlterPartitionManager.class);
+ private final NodeToControllerChannelManager controllerChannelManager;
+ private final Scheduler scheduler;
+ private final int brokerId;
+ private final Supplier<Long> brokerEpochSupplier;
+
+ // Used to allow only one pending ISR update per partition (visible for testing)
+ final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem> unsentIsrUpdates = new ConcurrentHashMap<>();
+
+ // Used to allow only one in-flight request at a time
+ private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
+
+ private DefaultAlterPartitionManager(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ this.controllerChannelManager = controllerChannelManager;
+ this.scheduler = scheduler;
+ this.brokerId = brokerId;
+ this.brokerEpochSupplier = brokerEpochSupplier;
+ }
+
+ // For testing
+ public static DefaultAlterPartitionManager create(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, Time time, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ return new DefaultAlterPartitionManager(controllerChannelManager, scheduler, brokerId, brokerEpochSupplier);
+ }
+
+ public static DefaultAlterPartitionManager create(AbstractKafkaConfig config,
+ Scheduler scheduler,
+ Supplier<ControllerInformation> controllerNodeProvider,
+ Time time,
+ Metrics metrics,
+ String threadNamePrefix,
+ Supplier<Long> brokerEpochSupplier) {
+ NodeToControllerChannelManager channelManager = new NodeToControllerChannelManagerImpl(
+ controllerNodeProvider,
+ time,
+ metrics,
+ config,
+ "alter-partition",
+ threadNamePrefix,
+ Long.MAX_VALUE
+ );
+
+ return new DefaultAlterPartitionManager(channelManager, scheduler, config.brokerId(), brokerEpochSupplier);
+ }
+
+ @Override
+ public void start() {
+ controllerChannelManager.start();
+ }
+
+ @Override
+ public void shutdown() throws InterruptedException {
+ controllerChannelManager.shutdown();
+ }
+
+ @Override
+ public CompletableFuture<LeaderAndIsr> submit(TopicIdPartition topicIdPartition,
+ LeaderAndIsr leaderAndIsr) {
+ CompletableFuture<LeaderAndIsr> future = new CompletableFuture<>();
+ AlterPartitionItem alterPartitionItem = new AlterPartitionItem(topicIdPartition, leaderAndIsr, future);
+ boolean enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition(), alterPartitionItem) == null;
+ if (enqueued) {
+ maybePropagateIsrChanges();
+ } else {
+ future.completeExceptionally(new OperationNotAttemptedException(String.format(
+ "Failed to enqueue ISR change state %s for partition %s", leaderAndIsr, topicIdPartition)));
+ }
+ return future;
+ }
+
+ void maybePropagateIsrChanges() {
+ // Send all pending items if there is not already a request in-flight.
+ if (!unsentIsrUpdates.isEmpty() && inflightRequest.compareAndSet(false, true)) {
+ // Copy current unsent ISRs but don't remove from the map, they get cleared in the response handler
+ List<AlterPartitionItem> inflightAlterPartitionItems = new ArrayList<>(unsentIsrUpdates.values());
+ sendRequest(inflightAlterPartitionItems);
+ }
+ }
+
+ void clearInFlightRequest() {
+ if (!inflightRequest.compareAndSet(true, false)) {
+ log.warn("Attempting to clear AlterPartition in-flight flag when no apparent request is in-flight");
+ }
+ }
+
+ private void sendRequest(List<AlterPartitionItem> inflightAlterPartitionItems) {
+ long brokerEpoch = brokerEpochSupplier.get();
+ AlterPartitionRequest.Builder request = buildRequest(inflightAlterPartitionItems, brokerEpoch);
+ log.debug("Sending AlterPartition to controller {}", request);
+
+ // We will not time out AlterPartition request, instead letting it retry indefinitely
+ // until a response is received, or a new LeaderAndIsr overwrites the existing isrState
+ // which causes the response for those partitions to be ignored.
+ controllerChannelManager.sendRequest(request,
+ new ControllerRequestCompletionHandler() {
+ @Override
+ public void onComplete(ClientResponse response) {
+ log.debug("Received AlterPartition response {}", response);
+ Errors error;
+ try {
+ if (response.authenticationException() != null) {
+ // For now, we treat authentication errors as retriable. We use the
+ // `NETWORK_EXCEPTION` error code for lack of a good alternative.
+ // Note that `NodeToControllerChannelManager` will still log the
+ // authentication errors so that users have a chance to fix the problem.
+ error = Errors.NETWORK_EXCEPTION;
+ } else if (response.versionMismatch() != null) {
+ error = Errors.UNSUPPORTED_VERSION;
+ } else {
+ error = handleAlterPartitionResponse(
+ (AlterPartitionResponse) response.responseBody(),
+ brokerEpoch,
+ inflightAlterPartitionItems
+ );
+ }
+ } finally {
+ // clear the flag so future requests can proceed
+ clearInFlightRequest();
+ }
+
+ // check if we need to send another request right away
+ if (Objects.requireNonNull(error) == Errors.NONE) {
+ // In the normal case, check for pending updates to send immediately
+ maybePropagateIsrChanges();
+ } else {
+ // If we received a top-level error from the controller, retry the request in the near future
+ scheduler.scheduleOnce("send-alter-partition", () -> maybePropagateIsrChanges(), 50);
+ }
+ }
+
+ @Override
+ public void onTimeout() {
+ throw new IllegalStateException("Encountered unexpected timeout when sending AlterPartition to the controller");
+ }
+ });
+ }
+
+ /**
+ * Builds an AlterPartition request.
+ * <p>
+ * While building the request, we don't know which version of the AlterPartition API is
+ * supported by the controller. The final decision is taken when the AlterPartitionRequest
+ * is built in the network client based on the advertised api versions of the controller.
+ *
+ * @return an AlterPartitionRequest.Builder with the provided parameters.
+ */
+ private AlterPartitionRequest.Builder buildRequest(
+ List<AlterPartitionItem> inflightAlterPartitionItems,
+ long brokerEpoch) {
+ AlterPartitionRequestData message = new AlterPartitionRequestData()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(brokerEpoch);
+
+ inflightAlterPartitionItems.stream()
+ .collect(Collectors.groupingBy(item -> item.topicIdPartition().topicId()))
+ .forEach((topicId, items) -> {
+ AlterPartitionRequestData.TopicData topicData = new AlterPartitionRequestData.TopicData().setTopicId(topicId);
+ message.topics().add(topicData);
+
+ items.forEach(item -> {
+ AlterPartitionRequestData.PartitionData partitionData =
+ new AlterPartitionRequestData.PartitionData()
+ .setPartitionIndex(item.topicIdPartition().partitionId())
+ .setLeaderEpoch(item.leaderAndIsr().leaderEpoch())
+ .setNewIsrWithEpochs(item.leaderAndIsr().isrWithBrokerEpoch())
+ .setPartitionEpoch(item.leaderAndIsr().partitionEpoch());
+
+ partitionData.setLeaderRecoveryState(item.leaderAndIsr().leaderRecoveryState().value());
+
+ topicData.partitions().add(partitionData);
+ });
+
+ });
+
+ return new AlterPartitionRequest.Builder(message);
+ }
+
+ private Errors handleAlterPartitionResponse(AlterPartitionResponse alterPartitionResponse,
+ long sentBrokerEpoch,
+ List<AlterPartitionItem> inflightAlterPartitionItems) {
+ AlterPartitionResponseData data = alterPartitionResponse.data();
+ Errors e = Errors.forCode(data.errorCode());
Review Comment:
`e` -> `error`
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.ControllerInformation;
+import org.apache.kafka.server.NodeToControllerChannelManagerImpl;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.config.AbstractKafkaConfig;
+import org.apache.kafka.server.util.Scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class DefaultAlterPartitionManager implements AlterPartitionManager {
+ private static final Logger log = LoggerFactory.getLogger(DefaultAlterPartitionManager.class);
+ private final NodeToControllerChannelManager controllerChannelManager;
+ private final Scheduler scheduler;
+ private final int brokerId;
+ private final Supplier<Long> brokerEpochSupplier;
+
+ // Used to allow only one pending ISR update per partition (visible for testing)
+ final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem> unsentIsrUpdates = new ConcurrentHashMap<>();
+
+ // Used to allow only one in-flight request at a time
+ private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
+
+ private DefaultAlterPartitionManager(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ this.controllerChannelManager = controllerChannelManager;
+ this.scheduler = scheduler;
+ this.brokerId = brokerId;
+ this.brokerEpochSupplier = brokerEpochSupplier;
+ }
+
+ // For testing
+ public static DefaultAlterPartitionManager create(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, Time time, int brokerId, Supplier<Long> brokerEpochSupplier) {
Review Comment:
Why do we use a factory instead of the constructor?
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.ControllerInformation;
+import org.apache.kafka.server.NodeToControllerChannelManagerImpl;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.config.AbstractKafkaConfig;
+import org.apache.kafka.server.util.Scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class DefaultAlterPartitionManager implements AlterPartitionManager {
+ private static final Logger log = LoggerFactory.getLogger(DefaultAlterPartitionManager.class);
+ private final NodeToControllerChannelManager controllerChannelManager;
+ private final Scheduler scheduler;
+ private final int brokerId;
+ private final Supplier<Long> brokerEpochSupplier;
+
+ // Used to allow only one pending ISR update per partition (visible for testing)
+ final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem> unsentIsrUpdates = new ConcurrentHashMap<>();
+
+ // Used to allow only one in-flight request at a time
+ private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
+
+ private DefaultAlterPartitionManager(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ this.controllerChannelManager = controllerChannelManager;
+ this.scheduler = scheduler;
+ this.brokerId = brokerId;
+ this.brokerEpochSupplier = brokerEpochSupplier;
+ }
+
+ // For testing
+ public static DefaultAlterPartitionManager create(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, Time time, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ return new DefaultAlterPartitionManager(controllerChannelManager, scheduler, brokerId, brokerEpochSupplier);
+ }
+
+ public static DefaultAlterPartitionManager create(AbstractKafkaConfig config,
+ Scheduler scheduler,
+ Supplier<ControllerInformation> controllerNodeProvider,
+ Time time,
+ Metrics metrics,
+ String threadNamePrefix,
+ Supplier<Long> brokerEpochSupplier) {
+ NodeToControllerChannelManager channelManager = new NodeToControllerChannelManagerImpl(
+ controllerNodeProvider,
+ time,
+ metrics,
+ config,
+ "alter-partition",
+ threadNamePrefix,
+ Long.MAX_VALUE
+ );
+
+ return new DefaultAlterPartitionManager(channelManager, scheduler, config.brokerId(), brokerEpochSupplier);
+ }
+
+ @Override
+ public void start() {
+ controllerChannelManager.start();
+ }
+
+ @Override
+ public void shutdown() throws InterruptedException {
+ controllerChannelManager.shutdown();
+ }
+
+ @Override
+ public CompletableFuture<LeaderAndIsr> submit(TopicIdPartition topicIdPartition,
+ LeaderAndIsr leaderAndIsr) {
+ CompletableFuture<LeaderAndIsr> future = new CompletableFuture<>();
+ AlterPartitionItem alterPartitionItem = new AlterPartitionItem(topicIdPartition, leaderAndIsr, future);
+ boolean enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition(), alterPartitionItem) == null;
+ if (enqueued) {
+ maybePropagateIsrChanges();
+ } else {
+ future.completeExceptionally(new OperationNotAttemptedException(String.format(
+ "Failed to enqueue ISR change state %s for partition %s", leaderAndIsr, topicIdPartition)));
+ }
+ return future;
+ }
+
+ void maybePropagateIsrChanges() {
+ // Send all pending items if there is not already a request in-flight.
+ if (!unsentIsrUpdates.isEmpty() && inflightRequest.compareAndSet(false, true)) {
+ // Copy current unsent ISRs but don't remove from the map, they get cleared in the response handler
+ List<AlterPartitionItem> inflightAlterPartitionItems = new ArrayList<>(unsentIsrUpdates.values());
+ sendRequest(inflightAlterPartitionItems);
+ }
+ }
+
+ void clearInFlightRequest() {
+ if (!inflightRequest.compareAndSet(true, false)) {
+ log.warn("Attempting to clear AlterPartition in-flight flag when no apparent request is in-flight");
+ }
+ }
+
+ private void sendRequest(List<AlterPartitionItem> inflightAlterPartitionItems) {
+ long brokerEpoch = brokerEpochSupplier.get();
+ AlterPartitionRequest.Builder request = buildRequest(inflightAlterPartitionItems, brokerEpoch);
+ log.debug("Sending AlterPartition to controller {}", request);
+
+ // We will not time out AlterPartition request, instead letting it retry indefinitely
+ // until a response is received, or a new LeaderAndIsr overwrites the existing isrState
+ // which causes the response for those partitions to be ignored.
+ controllerChannelManager.sendRequest(request,
+ new ControllerRequestCompletionHandler() {
+ @Override
+ public void onComplete(ClientResponse response) {
+ log.debug("Received AlterPartition response {}", response);
+ Errors error;
+ try {
+ if (response.authenticationException() != null) {
+ // For now, we treat authentication errors as retriable. We use the
+ // `NETWORK_EXCEPTION` error code for lack of a good alternative.
+ // Note that `NodeToControllerChannelManager` will still log the
+ // authentication errors so that users have a chance to fix the problem.
+ error = Errors.NETWORK_EXCEPTION;
+ } else if (response.versionMismatch() != null) {
+ error = Errors.UNSUPPORTED_VERSION;
+ } else {
+ error = handleAlterPartitionResponse(
+ (AlterPartitionResponse) response.responseBody(),
+ brokerEpoch,
+ inflightAlterPartitionItems
+ );
+ }
+ } finally {
+ // clear the flag so future requests can proceed
+ clearInFlightRequest();
+ }
+
+ // check if we need to send another request right away
+ if (Objects.requireNonNull(error) == Errors.NONE) {
+ // In the normal case, check for pending updates to send immediately
+ maybePropagateIsrChanges();
+ } else {
+ // If we received a top-level error from the controller, retry the request in the near future
+ scheduler.scheduleOnce("send-alter-partition", () -> maybePropagateIsrChanges(), 50);
+ }
+ }
+
+ @Override
+ public void onTimeout() {
+ throw new IllegalStateException("Encountered unexpected timeout when sending AlterPartition to the controller");
+ }
+ });
+ }
+
+ /**
+ * Builds an AlterPartition request.
+ * <p>
+ * While building the request, we don't know which version of the AlterPartition API is
+ * supported by the controller. The final decision is taken when the AlterPartitionRequest
+ * is built in the network client based on the advertised api versions of the controller.
+ *
+ * @return an AlterPartitionRequest.Builder with the provided parameters.
+ */
+ private AlterPartitionRequest.Builder buildRequest(
+ List<AlterPartitionItem> inflightAlterPartitionItems,
+ long brokerEpoch) {
+ AlterPartitionRequestData message = new AlterPartitionRequestData()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(brokerEpoch);
+
+ inflightAlterPartitionItems.stream()
+ .collect(Collectors.groupingBy(item -> item.topicIdPartition().topicId()))
+ .forEach((topicId, items) -> {
+ AlterPartitionRequestData.TopicData topicData = new AlterPartitionRequestData.TopicData().setTopicId(topicId);
+ message.topics().add(topicData);
+
+ items.forEach(item -> {
+ AlterPartitionRequestData.PartitionData partitionData =
+ new AlterPartitionRequestData.PartitionData()
+ .setPartitionIndex(item.topicIdPartition().partitionId())
+ .setLeaderEpoch(item.leaderAndIsr().leaderEpoch())
+ .setNewIsrWithEpochs(item.leaderAndIsr().isrWithBrokerEpoch())
+ .setPartitionEpoch(item.leaderAndIsr().partitionEpoch());
+
+ partitionData.setLeaderRecoveryState(item.leaderAndIsr().leaderRecoveryState().value());
+
+ topicData.partitions().add(partitionData);
+ });
+
+ });
+
+ return new AlterPartitionRequest.Builder(message);
+ }
+
+ private Errors handleAlterPartitionResponse(AlterPartitionResponse alterPartitionResponse,
+ long sentBrokerEpoch,
+ List<AlterPartitionItem> inflightAlterPartitionItems) {
+ AlterPartitionResponseData data = alterPartitionResponse.data();
+ Errors e = Errors.forCode(data.errorCode());
+ switch (e) {
+ case STALE_BROKER_EPOCH -> log.warn("Broker had a stale broker epoch ({}), retrying.", sentBrokerEpoch);
+ case CLUSTER_AUTHORIZATION_FAILED -> log.error("Broker is not authorized to send AlterPartition to controller",
+ Errors.CLUSTER_AUTHORIZATION_FAILED.exception("Broker is not authorized to send AlterPartition to controller"));
+ case NONE -> {
+ // Collect partition-level responses to pass to the callbacks
+ Map<TopicIdPartition, LeaderAndIsr> successResponses = new HashMap<>();
+ Map<TopicIdPartition, Errors> errorResponses = new HashMap<>();
+ data.topics().forEach(topic ->
+ topic.partitions().forEach(partition -> {
+ TopicIdPartition tp = new TopicIdPartition(topic.topicId(), partition.partitionIndex());
+ Errors apiError = Errors.forCode(partition.errorCode());
+ log.debug("Controller successfully handled AlterPartition request for {}: {}", tp, partition);
+ if (apiError == Errors.NONE) {
+ Optional<LeaderRecoveryState> leaderRecoveryStateOpt =
+ LeaderRecoveryState.optionalOf(partition.leaderRecoveryState());
+ if (leaderRecoveryStateOpt.isPresent()) {
+ successResponses.put(tp, new LeaderAndIsr(
+ partition.leaderId(),
+ partition.leaderEpoch(),
+ partition.isr(),
+ leaderRecoveryStateOpt.get(),
+ partition.partitionEpoch()
+ ));
+ } else {
+ log.error("Controller returned an invalid leader recovery state ({}) for {}: {}", partition.leaderRecoveryState(), tp, partition);
+ errorResponses.put(tp, Errors.UNKNOWN_SERVER_ERROR);
+ }
+ } else {
+ errorResponses.put(tp, apiError);
+ }
+ }
+ ));
+ // Iterate across the items we sent rather than what we received to ensure we run the callback even if a
+ // partition was somehow erroneously excluded from the response. Note that these callbacks are run from
+ // the leaderIsrUpdateLock write lock in Partition#sendAlterPartitionRequest
+ inflightAlterPartitionItems.forEach(inflightAlterPartition -> {
+ if (successResponses.get(inflightAlterPartition.topicIdPartition()) != null) {
Review Comment:
Can we use `containsKey()` instead? Same below
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19914: Move AlterPartitionManager to server module [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21432:
URL: https://github.com/apache/kafka/pull/21432#discussion_r3065037146
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -0,0 +1,287 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.ControllerInformation;
+import org.apache.kafka.server.NodeToControllerChannelManagerImpl;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.config.AbstractKafkaConfig;
+import org.apache.kafka.server.util.Scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class DefaultAlterPartitionManager implements AlterPartitionManager {
+ private static final Logger log = LoggerFactory.getLogger(DefaultAlterPartitionManager.class);
+ private final NodeToControllerChannelManager controllerChannelManager;
+ private final Scheduler scheduler;
+ private final int brokerId;
+ private final Supplier<Long> brokerEpochSupplier;
+
+ // Used to allow only one pending ISR update per partition (visible for testing)
+ final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem> unsentIsrUpdates = new ConcurrentHashMap<>();
+
+ // Used to allow only one in-flight request at a time
+ private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
+
+ public DefaultAlterPartitionManager(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ this.controllerChannelManager = controllerChannelManager;
+ this.scheduler = scheduler;
+ this.brokerId = brokerId;
+ this.brokerEpochSupplier = brokerEpochSupplier;
+ }
+
+ public static DefaultAlterPartitionManager create(AbstractKafkaConfig config,
+ Scheduler scheduler,
+ Supplier<ControllerInformation> controllerNodeProvider,
+ Time time,
+ Metrics metrics,
+ String threadNamePrefix,
+ Supplier<Long> brokerEpochSupplier) {
+ NodeToControllerChannelManager channelManager = new NodeToControllerChannelManagerImpl(
+ controllerNodeProvider,
+ time,
+ metrics,
+ config,
+ "alter-partition",
+ threadNamePrefix,
+ Long.MAX_VALUE
+ );
+
+ return new DefaultAlterPartitionManager(channelManager, scheduler, config.brokerId(), brokerEpochSupplier);
+ }
+
+ @Override
+ public void start() {
+ controllerChannelManager.start();
+ }
+
+ @Override
+ public void shutdown() throws InterruptedException {
+ controllerChannelManager.shutdown();
+ }
+
+ @Override
+ public CompletableFuture<LeaderAndIsr> submit(TopicIdPartition topicIdPartition,
+ LeaderAndIsr leaderAndIsr) {
+ CompletableFuture<LeaderAndIsr> future = new CompletableFuture<>();
+ AlterPartitionItem alterPartitionItem = new AlterPartitionItem(topicIdPartition, leaderAndIsr, future);
+ boolean enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition(), alterPartitionItem) == null;
+ if (enqueued) {
+ maybePropagateIsrChanges();
+ } else {
+ future.completeExceptionally(new OperationNotAttemptedException(String.format(
+ "Failed to enqueue ISR change state %s for partition %s", leaderAndIsr, topicIdPartition)));
+ }
+ return future;
+ }
+
+ void maybePropagateIsrChanges() {
+ // Send all pending items if there is not already a request in-flight.
+ if (!unsentIsrUpdates.isEmpty() && inflightRequest.compareAndSet(false, true)) {
+ // Copy current unsent ISRs but don't remove from the map, they get cleared in the response handler
+ List<AlterPartitionItem> inflightAlterPartitionItems = new ArrayList<>(unsentIsrUpdates.values());
+ sendRequest(inflightAlterPartitionItems);
+ }
+ }
+
+ void clearInFlightRequest() {
+ if (!inflightRequest.compareAndSet(true, false)) {
+ log.warn("Attempting to clear AlterPartition in-flight flag when no apparent request is in-flight");
+ }
+ }
+
+ private void sendRequest(List<AlterPartitionItem> inflightAlterPartitionItems) {
+ long brokerEpoch = brokerEpochSupplier.get();
+ AlterPartitionRequest.Builder request = buildRequest(inflightAlterPartitionItems, brokerEpoch);
+ log.debug("Sending AlterPartition to controller {}", request);
+
+ // We will not time out AlterPartition request, instead letting it retry indefinitely
+ // until a response is received, or a new LeaderAndIsr overwrites the existing isrState
+ // which causes the response for those partitions to be ignored.
+ controllerChannelManager.sendRequest(request,
+ new ControllerRequestCompletionHandler() {
+ @Override
+ public void onComplete(ClientResponse response) {
+ log.debug("Received AlterPartition response {}", response);
+ Errors error;
+ try {
+ if (response.authenticationException() != null) {
+ // For now, we treat authentication errors as retriable. We use the
+ // `NETWORK_EXCEPTION` error code for lack of a good alternative.
+ // Note that `NodeToControllerChannelManager` will still log the
+ // authentication errors so that users have a chance to fix the problem.
+ error = Errors.NETWORK_EXCEPTION;
+ } else if (response.versionMismatch() != null) {
+ error = Errors.UNSUPPORTED_VERSION;
+ } else {
+ error = handleAlterPartitionResponse(
+ (AlterPartitionResponse) response.responseBody(),
+ brokerEpoch,
+ inflightAlterPartitionItems
+ );
+ }
+ } finally {
+ // clear the flag so future requests can proceed
+ clearInFlightRequest();
+ }
+
+ // check if we need to send another request right away
+ if (error == Errors.NONE) {
+ // In the normal case, check for pending updates to send immediately
+ maybePropagateIsrChanges();
+ } else {
+ // If we received a top-level error from the controller, retry the request in the near future
+ scheduler.scheduleOnce("send-alter-partition", () -> maybePropagateIsrChanges(), 50);
+ }
+ }
+
+ @Override
+ public void onTimeout() {
+ throw new IllegalStateException("Encountered unexpected timeout when sending AlterPartition to the controller");
+ }
+ });
+ }
+
+ /**
+ * Builds an AlterPartition request.
+ * <p>
+ * While building the request, we don't know which version of the AlterPartition API is
+ * supported by the controller. The final decision is taken when the AlterPartitionRequest
+ * is built in the network client based on the advertised api versions of the controller.
+ *
+ * @return an AlterPartitionRequest.Builder with the provided parameters.
+ */
+ private AlterPartitionRequest.Builder buildRequest(
+ List<AlterPartitionItem> inflightAlterPartitionItems,
+ long brokerEpoch) {
+ AlterPartitionRequestData message = new AlterPartitionRequestData()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(brokerEpoch);
+
+ inflightAlterPartitionItems.stream()
+ .collect(Collectors.groupingBy(item -> item.topicIdPartition().topicId()))
+ .forEach((topicId, items) -> {
+ AlterPartitionRequestData.TopicData topicData = new AlterPartitionRequestData.TopicData().setTopicId(topicId);
+ message.topics().add(topicData);
+
+ items.forEach(item -> {
+ AlterPartitionRequestData.PartitionData partitionData =
+ new AlterPartitionRequestData.PartitionData()
+ .setPartitionIndex(item.topicIdPartition().partitionId())
+ .setLeaderEpoch(item.leaderAndIsr().leaderEpoch())
+ .setNewIsrWithEpochs(item.leaderAndIsr().isrWithBrokerEpoch())
+ .setPartitionEpoch(item.leaderAndIsr().partitionEpoch());
+
+ partitionData.setLeaderRecoveryState(item.leaderAndIsr().leaderRecoveryState().value());
+
+ topicData.partitions().add(partitionData);
+ });
+
+ });
+
+ return new AlterPartitionRequest.Builder(message);
+ }
+
+ private Errors handleAlterPartitionResponse(AlterPartitionResponse alterPartitionResponse,
+ long sentBrokerEpoch,
+ List<AlterPartitionItem> inflightAlterPartitionItems) {
+ AlterPartitionResponseData data = alterPartitionResponse.data();
+ Errors error = Errors.forCode(data.errorCode());
+ switch (error) {
+ case STALE_BROKER_EPOCH -> log.warn("Broker had a stale broker epoch ({}), retrying.", sentBrokerEpoch);
+ case CLUSTER_AUTHORIZATION_FAILED -> log.error("Broker is not authorized to send AlterPartition to controller",
Review Comment:
```java
case CLUSTER_AUTHORIZATION_FAILED -> {
String msg = "Broker is not authorized to send AlterPartition to controller";
log.error(msg, Errors.CLUSTER_AUTHORIZATION_FAILED.exception(msg));
}```
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -0,0 +1,287 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.ControllerInformation;
+import org.apache.kafka.server.NodeToControllerChannelManagerImpl;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.config.AbstractKafkaConfig;
+import org.apache.kafka.server.util.Scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class DefaultAlterPartitionManager implements AlterPartitionManager {
+ private static final Logger log = LoggerFactory.getLogger(DefaultAlterPartitionManager.class);
+ private final NodeToControllerChannelManager controllerChannelManager;
+ private final Scheduler scheduler;
+ private final int brokerId;
+ private final Supplier<Long> brokerEpochSupplier;
+
+ // Used to allow only one pending ISR update per partition (visible for testing)
+ final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem> unsentIsrUpdates = new ConcurrentHashMap<>();
+
+ // Used to allow only one in-flight request at a time
+ private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
+
+ public DefaultAlterPartitionManager(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ this.controllerChannelManager = controllerChannelManager;
+ this.scheduler = scheduler;
+ this.brokerId = brokerId;
+ this.brokerEpochSupplier = brokerEpochSupplier;
+ }
+
+ public static DefaultAlterPartitionManager create(AbstractKafkaConfig config,
+ Scheduler scheduler,
+ Supplier<ControllerInformation> controllerNodeProvider,
+ Time time,
+ Metrics metrics,
+ String threadNamePrefix,
+ Supplier<Long> brokerEpochSupplier) {
+ NodeToControllerChannelManager channelManager = new NodeToControllerChannelManagerImpl(
+ controllerNodeProvider,
+ time,
+ metrics,
+ config,
+ "alter-partition",
+ threadNamePrefix,
+ Long.MAX_VALUE
+ );
+
+ return new DefaultAlterPartitionManager(channelManager, scheduler, config.brokerId(), brokerEpochSupplier);
+ }
+
+ @Override
+ public void start() {
+ controllerChannelManager.start();
+ }
+
+ @Override
+ public void shutdown() throws InterruptedException {
+ controllerChannelManager.shutdown();
+ }
+
+ @Override
+ public CompletableFuture<LeaderAndIsr> submit(TopicIdPartition topicIdPartition,
+ LeaderAndIsr leaderAndIsr) {
+ CompletableFuture<LeaderAndIsr> future = new CompletableFuture<>();
+ AlterPartitionItem alterPartitionItem = new AlterPartitionItem(topicIdPartition, leaderAndIsr, future);
+ boolean enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition(), alterPartitionItem) == null;
+ if (enqueued) {
+ maybePropagateIsrChanges();
+ } else {
+ future.completeExceptionally(new OperationNotAttemptedException(String.format(
+ "Failed to enqueue ISR change state %s for partition %s", leaderAndIsr, topicIdPartition)));
+ }
+ return future;
+ }
+
+ void maybePropagateIsrChanges() {
+ // Send all pending items if there is not already a request in-flight.
+ if (!unsentIsrUpdates.isEmpty() && inflightRequest.compareAndSet(false, true)) {
+ // Copy current unsent ISRs but don't remove from the map, they get cleared in the response handler
+ List<AlterPartitionItem> inflightAlterPartitionItems = new ArrayList<>(unsentIsrUpdates.values());
+ sendRequest(inflightAlterPartitionItems);
+ }
+ }
+
+ void clearInFlightRequest() {
+ if (!inflightRequest.compareAndSet(true, false)) {
+ log.warn("Attempting to clear AlterPartition in-flight flag when no apparent request is in-flight");
+ }
+ }
+
+ private void sendRequest(List<AlterPartitionItem> inflightAlterPartitionItems) {
+ long brokerEpoch = brokerEpochSupplier.get();
+ AlterPartitionRequest.Builder request = buildRequest(inflightAlterPartitionItems, brokerEpoch);
+ log.debug("Sending AlterPartition to controller {}", request);
+
+ // We will not time out AlterPartition request, instead letting it retry indefinitely
+ // until a response is received, or a new LeaderAndIsr overwrites the existing isrState
+ // which causes the response for those partitions to be ignored.
+ controllerChannelManager.sendRequest(request,
+ new ControllerRequestCompletionHandler() {
+ @Override
+ public void onComplete(ClientResponse response) {
+ log.debug("Received AlterPartition response {}", response);
+ Errors error;
+ try {
+ if (response.authenticationException() != null) {
+ // For now, we treat authentication errors as retriable. We use the
+ // `NETWORK_EXCEPTION` error code for lack of a good alternative.
+ // Note that `NodeToControllerChannelManager` will still log the
+ // authentication errors so that users have a chance to fix the problem.
+ error = Errors.NETWORK_EXCEPTION;
+ } else if (response.versionMismatch() != null) {
+ error = Errors.UNSUPPORTED_VERSION;
+ } else {
+ error = handleAlterPartitionResponse(
+ (AlterPartitionResponse) response.responseBody(),
+ brokerEpoch,
+ inflightAlterPartitionItems
+ );
+ }
+ } finally {
+ // clear the flag so future requests can proceed
+ clearInFlightRequest();
+ }
+
+ // check if we need to send another request right away
+ if (error == Errors.NONE) {
+ // In the normal case, check for pending updates to send immediately
+ maybePropagateIsrChanges();
+ } else {
+ // If we received a top-level error from the controller, retry the request in the near future
+ scheduler.scheduleOnce("send-alter-partition", () -> maybePropagateIsrChanges(), 50);
+ }
+ }
+
+ @Override
+ public void onTimeout() {
+ throw new IllegalStateException("Encountered unexpected timeout when sending AlterPartition to the controller");
+ }
+ });
+ }
+
+ /**
+ * Builds an AlterPartition request.
+ * <p>
+ * While building the request, we don't know which version of the AlterPartition API is
+ * supported by the controller. The final decision is taken when the AlterPartitionRequest
+ * is built in the network client based on the advertised api versions of the controller.
+ *
+ * @return an AlterPartitionRequest.Builder with the provided parameters.
+ */
+ private AlterPartitionRequest.Builder buildRequest(
+ List<AlterPartitionItem> inflightAlterPartitionItems,
+ long brokerEpoch) {
+ AlterPartitionRequestData message = new AlterPartitionRequestData()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(brokerEpoch);
+
+ inflightAlterPartitionItems.stream()
+ .collect(Collectors.groupingBy(item -> item.topicIdPartition().topicId()))
+ .forEach((topicId, items) -> {
+ AlterPartitionRequestData.TopicData topicData = new AlterPartitionRequestData.TopicData().setTopicId(topicId);
+ message.topics().add(topicData);
+
+ items.forEach(item -> {
+ AlterPartitionRequestData.PartitionData partitionData =
+ new AlterPartitionRequestData.PartitionData()
+ .setPartitionIndex(item.topicIdPartition().partitionId())
+ .setLeaderEpoch(item.leaderAndIsr().leaderEpoch())
+ .setNewIsrWithEpochs(item.leaderAndIsr().isrWithBrokerEpoch())
+ .setPartitionEpoch(item.leaderAndIsr().partitionEpoch());
+
+ partitionData.setLeaderRecoveryState(item.leaderAndIsr().leaderRecoveryState().value());
+
+ topicData.partitions().add(partitionData);
+ });
+
+ });
+
+ return new AlterPartitionRequest.Builder(message);
+ }
+
+ private Errors handleAlterPartitionResponse(AlterPartitionResponse alterPartitionResponse,
+ long sentBrokerEpoch,
+ List<AlterPartitionItem> inflightAlterPartitionItems) {
+ AlterPartitionResponseData data = alterPartitionResponse.data();
+ Errors error = Errors.forCode(data.errorCode());
+ switch (error) {
+ case STALE_BROKER_EPOCH -> log.warn("Broker had a stale broker epoch ({}), retrying.", sentBrokerEpoch);
+ case CLUSTER_AUTHORIZATION_FAILED -> log.error("Broker is not authorized to send AlterPartition to controller",
Review Comment:
```java
case CLUSTER_AUTHORIZATION_FAILED -> {
String msg = "Broker is not authorized to send AlterPartition to controller";
log.error(msg, Errors.CLUSTER_AUTHORIZATION_FAILED.exception(msg));
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19914: Move AlterPartitionManager to server module [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21432:
URL: https://github.com/apache/kafka/pull/21432#issuecomment-3912083246
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19914: Move AlterPartitionManager to server module [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21432:
URL: https://github.com/apache/kafka/pull/21432#discussion_r3064624157
##########
build.gradle:
##########
@@ -3439,6 +3439,7 @@ project(':jmh-benchmarks') {
implementation project(':connect:json')
implementation project(':clients').sourceSets.test.output
implementation project(':server-common').sourceSets.test.output
+ implementation project(':server').sourceSets.test.output
Review Comment:
Is this required? I remove this change and the build still works
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -0,0 +1,287 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.ControllerInformation;
+import org.apache.kafka.server.NodeToControllerChannelManagerImpl;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.config.AbstractKafkaConfig;
+import org.apache.kafka.server.util.Scheduler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class DefaultAlterPartitionManager implements AlterPartitionManager {
+ private static final Logger log = LoggerFactory.getLogger(DefaultAlterPartitionManager.class);
+ private final NodeToControllerChannelManager controllerChannelManager;
+ private final Scheduler scheduler;
+ private final int brokerId;
+ private final Supplier<Long> brokerEpochSupplier;
+
+ // Used to allow only one pending ISR update per partition (visible for testing)
+ final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem> unsentIsrUpdates = new ConcurrentHashMap<>();
+
+ // Used to allow only one in-flight request at a time
+ private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
+
+ public DefaultAlterPartitionManager(NodeToControllerChannelManager controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long> brokerEpochSupplier) {
+ this.controllerChannelManager = controllerChannelManager;
+ this.scheduler = scheduler;
+ this.brokerId = brokerId;
+ this.brokerEpochSupplier = brokerEpochSupplier;
+ }
+
+ public static DefaultAlterPartitionManager create(AbstractKafkaConfig config,
+ Scheduler scheduler,
+ Supplier<ControllerInformation> controllerNodeProvider,
+ Time time,
+ Metrics metrics,
+ String threadNamePrefix,
+ Supplier<Long> brokerEpochSupplier) {
+ NodeToControllerChannelManager channelManager = new NodeToControllerChannelManagerImpl(
+ controllerNodeProvider,
+ time,
+ metrics,
+ config,
+ "alter-partition",
+ threadNamePrefix,
+ Long.MAX_VALUE
+ );
+
+ return new DefaultAlterPartitionManager(channelManager, scheduler, config.brokerId(), brokerEpochSupplier);
+ }
+
+ @Override
+ public void start() {
+ controllerChannelManager.start();
+ }
+
+ @Override
+ public void shutdown() throws InterruptedException {
+ controllerChannelManager.shutdown();
+ }
+
+ @Override
+ public CompletableFuture<LeaderAndIsr> submit(TopicIdPartition topicIdPartition,
+ LeaderAndIsr leaderAndIsr) {
+ CompletableFuture<LeaderAndIsr> future = new CompletableFuture<>();
+ AlterPartitionItem alterPartitionItem = new AlterPartitionItem(topicIdPartition, leaderAndIsr, future);
+ boolean enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition(), alterPartitionItem) == null;
+ if (enqueued) {
+ maybePropagateIsrChanges();
+ } else {
+ future.completeExceptionally(new OperationNotAttemptedException(String.format(
+ "Failed to enqueue ISR change state %s for partition %s", leaderAndIsr, topicIdPartition)));
+ }
+ return future;
+ }
+
+ void maybePropagateIsrChanges() {
+ // Send all pending items if there is not already a request in-flight.
+ if (!unsentIsrUpdates.isEmpty() && inflightRequest.compareAndSet(false, true)) {
+ // Copy current unsent ISRs but don't remove from the map, they get cleared in the response handler
+ List<AlterPartitionItem> inflightAlterPartitionItems = new ArrayList<>(unsentIsrUpdates.values());
+ sendRequest(inflightAlterPartitionItems);
+ }
+ }
+
+ void clearInFlightRequest() {
+ if (!inflightRequest.compareAndSet(true, false)) {
+ log.warn("Attempting to clear AlterPartition in-flight flag when no apparent request is in-flight");
+ }
+ }
+
+ private void sendRequest(List<AlterPartitionItem> inflightAlterPartitionItems) {
+ long brokerEpoch = brokerEpochSupplier.get();
+ AlterPartitionRequest.Builder request = buildRequest(inflightAlterPartitionItems, brokerEpoch);
+ log.debug("Sending AlterPartition to controller {}", request);
+
+ // We will not time out AlterPartition request, instead letting it retry indefinitely
+ // until a response is received, or a new LeaderAndIsr overwrites the existing isrState
+ // which causes the response for those partitions to be ignored.
+ controllerChannelManager.sendRequest(request,
+ new ControllerRequestCompletionHandler() {
+ @Override
+ public void onComplete(ClientResponse response) {
+ log.debug("Received AlterPartition response {}", response);
+ Errors error;
+ try {
+ if (response.authenticationException() != null) {
+ // For now, we treat authentication errors as retriable. We use the
+ // `NETWORK_EXCEPTION` error code for lack of a good alternative.
+ // Note that `NodeToControllerChannelManager` will still log the
+ // authentication errors so that users have a chance to fix the problem.
+ error = Errors.NETWORK_EXCEPTION;
+ } else if (response.versionMismatch() != null) {
+ error = Errors.UNSUPPORTED_VERSION;
+ } else {
+ error = handleAlterPartitionResponse(
+ (AlterPartitionResponse) response.responseBody(),
+ brokerEpoch,
+ inflightAlterPartitionItems
+ );
+ }
+ } finally {
+ // clear the flag so future requests can proceed
+ clearInFlightRequest();
+ }
+
+ // check if we need to send another request right away
+ if (error == Errors.NONE) {
+ // In the normal case, check for pending updates to send immediately
+ maybePropagateIsrChanges();
+ } else {
+ // If we received a top-level error from the controller, retry the request in the near future
+ scheduler.scheduleOnce("send-alter-partition", () -> maybePropagateIsrChanges(), 50);
+ }
+ }
+
+ @Override
+ public void onTimeout() {
+ throw new IllegalStateException("Encountered unexpected timeout when sending AlterPartition to the controller");
+ }
+ });
+ }
+
+ /**
+ * Builds an AlterPartition request.
+ * <p>
+ * While building the request, we don't know which version of the AlterPartition API is
+ * supported by the controller. The final decision is taken when the AlterPartitionRequest
+ * is built in the network client based on the advertised api versions of the controller.
+ *
+ * @return an AlterPartitionRequest.Builder with the provided parameters.
+ */
+ private AlterPartitionRequest.Builder buildRequest(
+ List<AlterPartitionItem> inflightAlterPartitionItems,
+ long brokerEpoch) {
+ AlterPartitionRequestData message = new AlterPartitionRequestData()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(brokerEpoch);
+
+ inflightAlterPartitionItems.stream()
+ .collect(Collectors.groupingBy(item -> item.topicIdPartition().topicId()))
+ .forEach((topicId, items) -> {
+ AlterPartitionRequestData.TopicData topicData = new AlterPartitionRequestData.TopicData().setTopicId(topicId);
Review Comment:
Let's go with `mapKey` to avoid the `grouping` party
##########
server/src/test/java/org/apache/kafka/server/partition/AlterPartitionManagerTest.java:
##########
@@ -0,0 +1,543 @@
+/*
+ * 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.kafka.server.partition;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.AuthenticationException;
+import org.apache.kafka.common.errors.OperationNotAttemptedException;
+import org.apache.kafka.common.errors.UnknownServerException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.message.AlterPartitionRequestData;
+import org.apache.kafka.common.message.AlterPartitionRequestData.BrokerState;
+import org.apache.kafka.common.message.AlterPartitionResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AlterPartitionRequest;
+import org.apache.kafka.common.requests.AlterPartitionResponse;
+import org.apache.kafka.common.requests.RequestHeader;
+import org.apache.kafka.metadata.LeaderAndIsr;
+import org.apache.kafka.metadata.LeaderRecoveryState;
+import org.apache.kafka.server.common.ControllerRequestCompletionHandler;
+import org.apache.kafka.server.common.NodeToControllerChannelManager;
+import org.apache.kafka.server.common.TopicIdPartition;
+import org.apache.kafka.server.util.MockScheduler;
+import org.apache.kafka.server.util.MockTime;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.ArgumentMatcher;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+
+import static org.apache.kafka.test.TestUtils.assertFutureThrows;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class AlterPartitionManagerTest {
+ private final Uuid topicId = Uuid.randomUuid();
+ private final MockTime time = new MockTime();
+ private final int brokerId = 1;
+
+ private NodeToControllerChannelManager brokerToController;
+
+ private final TopicIdPartition tp0 = new TopicIdPartition(topicId, 0);
+ private final TopicIdPartition tp1 = new TopicIdPartition(topicId, 1);
+ private final TopicIdPartition tp2 = new TopicIdPartition(topicId, 2);
+
+ @BeforeEach
+ public void setup() {
+ brokerToController = mock(NodeToControllerChannelManager.class);
+ }
+
+ @Test
+ public void testBasic() {
+ var scheduler = new MockScheduler(time);
+ var alterPartitionManager = new DefaultAlterPartitionManager(brokerToController, scheduler, brokerId, () -> 2L);
+ alterPartitionManager.start();
+ alterPartitionManager.submit(tp0, new LeaderAndIsr(1, 1, List.of(1, 2, 3), LeaderRecoveryState.RECOVERED, 10));
+ verify(brokerToController).start();
+ verify(brokerToController).sendRequest(any(), any());
+ }
+
+ @Test
+ public void testBasicWithBrokerEpoch() {
+ var scheduler = new MockScheduler(time);
+ var alterPartitionManager = new DefaultAlterPartitionManager(brokerToController, scheduler, brokerId, () -> 101L);
+ alterPartitionManager.start();
+ ArrayList<BrokerState> isrWithBrokerEpoch = new ArrayList<>();
+ for (int i = 1; i <= 3; i++) {
+ isrWithBrokerEpoch.add(new BrokerState().setBrokerId(i).setBrokerEpoch(100 + i));
+ }
+ alterPartitionManager.submit(tp0, new LeaderAndIsr(1, 1, LeaderRecoveryState.RECOVERED, isrWithBrokerEpoch, 10));
+
+ var expectedAlterPartitionData = new AlterPartitionRequestData()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(101);
+ var topicData = new AlterPartitionRequestData.TopicData().setTopicId(topicId);
+
+ ArrayList<BrokerState> newIsrWithBrokerEpoch = new ArrayList<>();
+ newIsrWithBrokerEpoch.add(new BrokerState().setBrokerId(1).setBrokerEpoch(101));
+ newIsrWithBrokerEpoch.add(new BrokerState().setBrokerId(2).setBrokerEpoch(102));
+ newIsrWithBrokerEpoch.add(new BrokerState().setBrokerId(3).setBrokerEpoch(103));
+
+ topicData.partitions().add(new AlterPartitionRequestData.PartitionData()
+ .setPartitionIndex(0)
+ .setLeaderEpoch(1)
+ .setPartitionEpoch(10)
+ .setNewIsrWithEpochs(newIsrWithBrokerEpoch));
+
+ expectedAlterPartitionData.topics().add(topicData);
+
+ verify(brokerToController).start();
+ ArgumentCaptor<AbstractRequest.Builder<AlterPartitionRequest>> captor = ArgumentCaptor.captor();
+ verify(brokerToController).sendRequest(captor.capture(), any());
+ AlterPartitionRequest.Builder builder = (AlterPartitionRequest.Builder) captor.getValue();
+ assertEquals(expectedAlterPartitionData, builder.build().data());
+ }
+
+ @ParameterizedTest
+ @EnumSource(LeaderRecoveryState.class)
+ public void testBasicSentLeaderRecoveryState(LeaderRecoveryState leaderRecoveryState) {
+ ArgumentCaptor<AbstractRequest.Builder<AlterPartitionRequest>> requestCapture = ArgumentCaptor.captor();
+
+ var scheduler = new MockScheduler(time);
+ var alterPartitionManager = new DefaultAlterPartitionManager(brokerToController, scheduler, brokerId, () -> 2L);
+ alterPartitionManager.start();
+ alterPartitionManager.submit(tp0, new LeaderAndIsr(1, 1, List.of(1), leaderRecoveryState, 10));
+ verify(brokerToController).start();
+ verify(brokerToController).sendRequest(requestCapture.capture(), any());
+
+ var request = requestCapture.getValue().build();
+ assertEquals(leaderRecoveryState.value(), request.data().topics().get(0).partitions().get(0).leaderRecoveryState());
+ }
+
+ @Test
+ public void testOverwriteWithinBatch() {
+ ArgumentCaptor<AbstractRequest.Builder<AlterPartitionRequest>> capture = ArgumentCaptor.captor();
+ ArgumentCaptor<ControllerRequestCompletionHandler> callbackCapture = ArgumentCaptor.captor();
+
+ var scheduler = new MockScheduler(time);
+ var alterPartitionManager = new DefaultAlterPartitionManager(brokerToController, scheduler, brokerId, () -> 2L);
+ alterPartitionManager.start();
+
+ // Only send one ISR update for a given topic+partition
+ var firstSubmitFuture = alterPartitionManager.submit(tp0, new LeaderAndIsr(1, 1, List.of(1, 2, 3), LeaderRecoveryState.RECOVERED, 10));
+ assertFalse(firstSubmitFuture.isDone());
+
+ var failedSubmitFuture = alterPartitionManager.submit(tp0, new LeaderAndIsr(1, 1, List.of(1, 2), LeaderRecoveryState.RECOVERED, 10));
+ assertTrue(failedSubmitFuture.isCompletedExceptionally());
+ assertFutureThrows(OperationNotAttemptedException.class, failedSubmitFuture);
+
+ // Simulate response
+ var alterPartitionResp = partitionResponse(tp0, Errors.NONE, 0, 0, 0, List.of());
+ var resp = makeClientResponse(alterPartitionResp, ApiKeys.ALTER_PARTITION.latestVersion());
+
+ verify(brokerToController).sendRequest(capture.capture(), callbackCapture.capture());
+ callbackCapture.getValue().onComplete(resp);
+
+ // Now we can submit this partition again
+ var newSubmitFuture = alterPartitionManager.submit(tp0, new LeaderAndIsr(1, 1, List.of(1), LeaderRecoveryState.RECOVERED, 10));
+ assertFalse(newSubmitFuture.isDone());
+
+ verify(brokerToController).start();
+ verify(brokerToController, times(2)).sendRequest(capture.capture(), callbackCapture.capture());
+
+ // Make sure we sent the right request ISR={1}
+ var request = capture.getValue().build();
+ assertEquals(1, request.data().topics().size());
+ if (request.version() < 3) {
+ assertEquals(1, request.data().topics().get(0).partitions().get(0).newIsr().size());
+ } else {
+ assertEquals(1, request.data().topics().get(0).partitions().get(0).newIsrWithEpochs().size());
+ }
+ }
+
+ @Test
+ public void testSingleBatch() {
+ ArgumentCaptor<AbstractRequest.Builder<AlterPartitionRequest>> capture = ArgumentCaptor.captor();
+ ArgumentCaptor<ControllerRequestCompletionHandler> callbackCapture = ArgumentCaptor.captor();
+
+ var scheduler = new MockScheduler(time);
+ var alterPartitionManager = new DefaultAlterPartitionManager(brokerToController, scheduler, brokerId, () -> 2L);
+ alterPartitionManager.start();
+
+ // First request will send batch of one
+ alterPartitionManager.submit(new TopicIdPartition(topicId, 0),
+ new LeaderAndIsr(1, 1, List.of(1, 2, 3), LeaderRecoveryState.RECOVERED, 10));
+
+ // Other submissions will queue up until a response
+ for (int i = 1; i <= 9; i++) {
+ alterPartitionManager.submit(new TopicIdPartition(topicId, i),
+ new LeaderAndIsr(1, 1, List.of(1, 2, 3), LeaderRecoveryState.RECOVERED, 10));
+ }
+
+ // Simulate response, omitting partition 0 will allow it to stay in unsent queue
+ var alterPartitionResp = new AlterPartitionResponse(new AlterPartitionResponseData());
+ var resp = new ClientResponse(null, null, "", 0L, 0L,
+ false, null, null, alterPartitionResp);
+
+ // On the callback, we check for unsent items and send another request
+ verify(brokerToController).sendRequest(capture.capture(), callbackCapture.capture());
+ callbackCapture.getValue().onComplete(resp);
+
+ verify(brokerToController).start();
+ verify(brokerToController, times(2)).sendRequest(capture.capture(), callbackCapture.capture());
+
+ // Verify the last request sent had all 10 items
+ var request = capture.getValue().build();
+ assertEquals(1, request.data().topics().size());
+ assertEquals(10, request.data().topics().get(0).partitions().size());
+ }
+
+ @Test
+ public void testSubmitFromCallback() throws ExecutionException, InterruptedException, TimeoutException {
+ // prepare a partition level retriable error response
+ var alterPartitionRespWithPartitionError = partitionResponse(tp0, Errors.UNKNOWN_SERVER_ERROR, 0, 0, 0, List.of());
+ var errorResponse = makeClientResponse(alterPartitionRespWithPartitionError, ApiKeys.ALTER_PARTITION.latestVersion());
+
+ var leaderId = 1;
+ var leaderEpoch = 1;
+ var partitionEpoch = 10;
+ var isr = List.of(1, 2, 3);
+ var leaderAndIsr = new LeaderAndIsr(leaderId, leaderEpoch, isr, LeaderRecoveryState.RECOVERED, partitionEpoch);
+ ArgumentCaptor<ControllerRequestCompletionHandler> callbackCapture = ArgumentCaptor.captor();
+
+ var scheduler = new MockScheduler(time);
+ var alterPartitionManager = new DefaultAlterPartitionManager(brokerToController, scheduler, brokerId, () -> 2L);
+ alterPartitionManager.start();
+ var future = alterPartitionManager.submit(tp0, leaderAndIsr);
+ var finalFuture = new CompletableFuture<LeaderAndIsr>();
+ future.whenComplete((result, error) -> {
+ if (error != null) {
+ // Retry when error.
+ alterPartitionManager.submit(tp0, leaderAndIsr).whenComplete((retryResult, retryError) -> {
+ if (retryError != null) {
+ finalFuture.completeExceptionally(retryError);
+ } else {
+ finalFuture.complete(retryResult);
+ }
+ });
+ } else {
+ finalFuture.completeExceptionally(new AssertionError("Expected to future to be failed"));
+ }
+ });
+
+ verify(brokerToController).start();
+ verify(brokerToController).sendRequest(any(), callbackCapture.capture());
+ reset(brokerToController);
+ callbackCapture.getValue().onComplete(errorResponse);
+
+ // Complete the retry request
+ var retryAlterPartitionResponse = partitionResponse(tp0, Errors.NONE, partitionEpoch, leaderId, leaderEpoch, isr);
+ var retryResponse = makeClientResponse(retryAlterPartitionResponse, ApiKeys.ALTER_PARTITION.latestVersion());
+
+ verify(brokerToController).sendRequest(any(), callbackCapture.capture());
+ callbackCapture.getValue().onComplete(retryResponse);
+
+ assertEquals(leaderAndIsr, finalFuture.get(200, TimeUnit.MILLISECONDS));
+ // No more items in unsentIsrUpdates
+ assertFalse(alterPartitionManager.unsentIsrUpdates.containsKey(tp0));
+
+ }
+
+ @Test
+ public void testAuthorizationFailed() {
+ testRetryOnTopLevelError(Errors.CLUSTER_AUTHORIZATION_FAILED);
+ }
+
+ @Test
+ public void testStaleBrokerEpoch() {
+ testRetryOnTopLevelError(Errors.STALE_BROKER_EPOCH);
+ }
+
+ @Test
+ public void testUnknownServer() {
+ testRetryOnTopLevelError(Errors.UNKNOWN_SERVER_ERROR);
+ }
+
+ @Test
+ public void testRetryOnAuthenticationFailure() {
+ testRetryOnErrorResponse(new ClientResponse(null, null, "", 0L, 0L,
+ false, null, new AuthenticationException("authentication failed"), null));
+ }
+
+ @Test
+ public void testRetryOnUnsupportedVersionError() {
+ testRetryOnErrorResponse(new ClientResponse(null, null, "", 0L, 0L,
+ false, new UnsupportedVersionException("unsupported version"), null, null));
+ }
+
+ private void testRetryOnTopLevelError(Errors error) {
+ var alterPartitionResp = new AlterPartitionResponse(new AlterPartitionResponseData().setErrorCode(error.code()));
+ var response = makeClientResponse(alterPartitionResp, ApiKeys.ALTER_PARTITION.latestVersion());
+ testRetryOnErrorResponse(response);
+ }
+
+ private void testRetryOnErrorResponse(ClientResponse response) {
+ var leaderAndIsr = new LeaderAndIsr(1, 1, List.of(1, 2, 3), LeaderRecoveryState.RECOVERED, 10);
+ ArgumentCaptor<ControllerRequestCompletionHandler> callbackCapture = ArgumentCaptor.captor();
+
+ var scheduler = new MockScheduler(time);
+ var alterPartitionManager = new DefaultAlterPartitionManager(brokerToController, scheduler, brokerId, () -> 2L);
+ alterPartitionManager.start();
+ alterPartitionManager.submit(tp0, leaderAndIsr);
+
+ verify(brokerToController).start();
+ verify(brokerToController).sendRequest(any(), callbackCapture.capture());
+ callbackCapture.getValue().onComplete(response);
+
+ // Any top-level error, we want to retry, so we don't clear items from the pending map
+ assertTrue(alterPartitionManager.unsentIsrUpdates.containsKey(tp0));
+
+ reset(brokerToController);
+
+ // After some time, we will retry failed requests
+ time.sleep(100);
+ scheduler.tick();
+
+ // After a successful response, we can submit another AlterIsrItem
+ var retryAlterPartitionResponse = partitionResponse(tp0, Errors.NONE, 0, 0, 0, List.of());
+ var retryResponse = makeClientResponse(retryAlterPartitionResponse, ApiKeys.ALTER_PARTITION.latestVersion());
+
+ verify(brokerToController).sendRequest(any(), callbackCapture.capture());
+ callbackCapture.getValue().onComplete(retryResponse);
+
+ assertFalse(alterPartitionManager.unsentIsrUpdates.containsKey(tp0));
+ }
+
+ @Test
+ public void testInvalidUpdateVersion() {
+ checkPartitionError(Errors.INVALID_UPDATE_VERSION);
+ }
+
+ @Test
+ public void testUnknownTopicPartition() {
+ checkPartitionError(Errors.UNKNOWN_TOPIC_OR_PARTITION);
+ }
+
+ @Test
+ public void testNotLeaderOrFollower() {
+ checkPartitionError(Errors.NOT_LEADER_OR_FOLLOWER);
+ }
+
+ @Test
+ public void testInvalidRequest() {
+ checkPartitionError(Errors.INVALID_REQUEST);
+ }
+
+ private void checkPartitionError(Errors error) {
+ var alterPartitionManager = testPartitionError(tp0, error);
+ var future = alterPartitionManager.submit(tp0, new LeaderAndIsr(1, 1, List.of(1, 2, 3), LeaderRecoveryState.RECOVERED, 10));
+ assertFalse(future.isDone());
+ }
+
+ private AlterPartitionManager testPartitionError(TopicIdPartition tp, Errors error) {
+ ArgumentCaptor<ControllerRequestCompletionHandler> callbackCapture = ArgumentCaptor.captor();
+ reset(brokerToController);
+
+ var scheduler = new MockScheduler(time);
+ var alterPartitionManager = new DefaultAlterPartitionManager(brokerToController, scheduler, brokerId, () -> 2L);
+ alterPartitionManager.start();
+
+ var future = alterPartitionManager.submit(tp, new LeaderAndIsr(1, 1, List.of(1, 2, 3), LeaderRecoveryState.RECOVERED, 10));
+
+ verify(brokerToController).start();
+ verify(brokerToController).sendRequest(any(), callbackCapture.capture());
+ reset(brokerToController);
+
+ var alterPartitionResp = partitionResponse(tp, error, 0, 0, 0, List.of());
+ var resp = makeClientResponse(alterPartitionResp, ApiKeys.ALTER_PARTITION.latestVersion());
+ callbackCapture.getValue().onComplete(resp);
+ assertTrue(future.isCompletedExceptionally());
+ assertFutureThrows(error.exception().getClass(), future);
+ return alterPartitionManager;
+ }
+
+ @Test
+ public void testOneInFlight() {
+ ArgumentCaptor<ControllerRequestCompletionHandler> callbackCapture = ArgumentCaptor.captor();
+
+ var scheduler = new MockScheduler(time);
+ var alterPartitionManager = new DefaultAlterPartitionManager(brokerToController, scheduler, brokerId, () -> 2L);
+ alterPartitionManager.start();
+
+ // First submit will send the request
+ alterPartitionManager.submit(tp0, new LeaderAndIsr(1, 1, List.of(1, 2, 3), LeaderRecoveryState.RECOVERED, 10));
+
+ // These will become pending unsent items
+ alterPartitionManager.submit(tp1, new LeaderAndIsr(1, 1, List.of(1, 2, 3), LeaderRecoveryState.RECOVERED, 10));
+ alterPartitionManager.submit(tp2, new LeaderAndIsr(1, 1, List.of(1, 2, 3), LeaderRecoveryState.RECOVERED, 10));
+
+ verify(brokerToController).start();
+ verify(brokerToController).sendRequest(any(), callbackCapture.capture());
+
+ // Once the callback runs, another request will be sent
+ reset(brokerToController);
+
+ var alterPartitionResp = new AlterPartitionResponse(new AlterPartitionResponseData());
+ var resp = makeClientResponse(alterPartitionResp, ApiKeys.ALTER_PARTITION.latestVersion());
+ callbackCapture.getValue().onComplete(resp);
+
+ // Verify that pending items (tp1, tp2) are sent after the callback
+ verify(brokerToController).sendRequest(any(), any());
+ }
+
+ @Test
+ public void testPartitionMissingInResponse() {
+ var expectedVersion = ApiKeys.ALTER_PARTITION.latestVersion();
+ var leaderAndIsr = new LeaderAndIsr(1, 1, List.of(1, 2, 3), LeaderRecoveryState.RECOVERED, 10);
+ var scheduler = new MockScheduler(time);
+ var alterPartitionManager = new DefaultAlterPartitionManager(brokerToController, scheduler, brokerId, () -> 2L);
+ alterPartitionManager.start();
+
+ // The first `submit` will send the `AlterIsr` request
+ var future1 = alterPartitionManager.submit(tp0, leaderAndIsr);
+ var callback1 = verifySendRequest(brokerToController, alterPartitionRequestMatcher(
+ Set.of(tp0),
+ expectedVersion
+ ));
+
+ // Additional calls while the `AlterIsr` request is inflight will be queued
+ var future2 = alterPartitionManager.submit(tp1, leaderAndIsr);
+ var future3 = alterPartitionManager.submit(tp2, leaderAndIsr);
+
+ // Respond to the first request, which will also allow the next request to get sent
+ callback1.onComplete(makeClientResponse(
+ partitionResponse(tp0, Errors.UNKNOWN_SERVER_ERROR, 0, 0, 0, List.of()),
+ expectedVersion
+ ));
+ assertFutureThrows(UnknownServerException.class, future1);
+ assertFalse(future2.isDone());
+ assertFalse(future3.isDone());
+
+ // Verify the second request includes both expected partitions, but only respond with one of them
+ var callback2 = verifySendRequest(brokerToController, alterPartitionRequestMatcher(
+ Set.of(tp1, tp2),
+ expectedVersion
+ ));
+ callback2.onComplete(makeClientResponse(
+ partitionResponse(tp2, Errors.UNKNOWN_SERVER_ERROR, 0, 0, 0, List.of()),
+ expectedVersion
+ ));
+ assertFutureThrows(UnknownServerException.class, future3);
+ assertFalse(future2.isDone());
+
+ // The missing partition should be retried
+ var callback3 = verifySendRequest(brokerToController, alterPartitionRequestMatcher(
+ Set.of(tp1),
+ expectedVersion
+ ));
+ callback3.onComplete(makeClientResponse(
+ partitionResponse(tp1, Errors.UNKNOWN_SERVER_ERROR, 0, 0, 0, List.of()),
+ expectedVersion
+ ));
+ assertFutureThrows(UnknownServerException.class, future2);
+
+ }
+
+ private ControllerRequestCompletionHandler verifySendRequest(NodeToControllerChannelManager brokerToController,
+ ArgumentMatcher<AbstractRequest.Builder<? extends AbstractRequest>> expectedRequest) {
+ ArgumentCaptor<ControllerRequestCompletionHandler> callbackCapture = ArgumentCaptor.captor();
+
+ verify(brokerToController).sendRequest(argThat(expectedRequest), callbackCapture.capture());
+
+ reset(brokerToController);
+
+ return callbackCapture.getValue();
+ }
+
+ private ArgumentMatcher<AbstractRequest.Builder<? extends AbstractRequest>> alterPartitionRequestMatcher(Set<TopicIdPartition> expectedTopicPartitions,
+ Short expectedVersion) {
+ return request -> {
+ assertEquals(ApiKeys.ALTER_PARTITION, request.apiKey());
+
+ var alterPartitionRequest = ((AlterPartitionRequest.Builder) request).build();
+ assertEquals(expectedVersion, alterPartitionRequest.version());
+
+ var requestTopicPartitions = alterPartitionRequest.data().topics().stream()
+ .flatMap(topicData ->
+ topicData.partitions().stream()
+ .map(partitionData ->
+ new TopicPartitionKey(topicData.topicId(), partitionData.partitionIndex()))
+ ).collect(Collectors.toSet());
+
+ var expectedSet = expectedTopicPartitions.stream()
+ .map(tp -> new TopicPartitionKey(tp.topicId(), tp.partitionId()))
+ .collect(Collectors.toSet());
+
+ return expectedSet.equals(requestTopicPartitions);
+ };
+ }
+
+ record TopicPartitionKey(Uuid topicId, int partitionId) { }
Review Comment:
this can be replaced by `TopicIdPartition`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org