You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by cm...@apache.org on 2024/08/08 23:14:27 UTC
(kafka) branch 3.9 updated: KAFKA-17223 Retrying the call after encoutering UnsupportedVersionException will cause ConcurrentModificationException (#16753)
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.9 by this push:
new c4cc6d2ff3b KAFKA-17223 Retrying the call after encoutering UnsupportedVersionException will cause ConcurrentModificationException (#16753)
c4cc6d2ff3b is described below
commit c4cc6d2ff3b043af0d818c3a6b2110ae2c07f221
Author: PoAn Yang <pa...@apache.org>
AuthorDate: Fri Aug 9 01:07:54 2024 +0800
KAFKA-17223 Retrying the call after encoutering UnsupportedVersionException will cause ConcurrentModificationException (#16753)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../kafka/clients/admin/KafkaAdminClient.java | 19 +++++++++---
.../kafka/clients/admin/KafkaAdminClientTest.java | 36 ++++++++++++++++++++++
2 files changed, 51 insertions(+), 4 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 3672b61647d..f1c52df5027 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -1197,16 +1197,27 @@ public class KafkaAdminClient extends AdminClient {
long pollTimeout = Long.MAX_VALUE;
log.trace("Trying to choose nodes for {} at {}", pendingCalls, now);
- Iterator<Call> pendingIter = pendingCalls.iterator();
- while (pendingIter.hasNext()) {
- Call call = pendingIter.next();
+ List<Call> toRemove = new ArrayList<>();
+ // Using pendingCalls.size() to get the list size before the for-loop to avoid infinite loop.
+ // If call.fail keeps adding the call to pendingCalls,
+ // the loop like for (int i = 0; i < pendingCalls.size(); i++) can't stop.
+ int pendingSize = pendingCalls.size();
+ // pendingCalls could be modified in this loop,
+ // hence using for-loop instead of iterator to avoid ConcurrentModificationException.
+ for (int i = 0; i < pendingSize; i++) {
+ Call call = pendingCalls.get(i);
// If the call is being retried, await the proper backoff before finding the node
if (now < call.nextAllowedTryMs) {
pollTimeout = Math.min(pollTimeout, call.nextAllowedTryMs - now);
} else if (maybeDrainPendingCall(call, now)) {
- pendingIter.remove();
+ toRemove.add(call);
}
}
+
+ // Use remove instead of removeAll to avoid delete all matched elements
+ for (Call call : toRemove) {
+ pendingCalls.remove(call);
+ }
return pollTimeout;
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
index dc74a178442..0f0dff65626 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
@@ -24,6 +24,7 @@ import org.apache.kafka.clients.MockClient;
import org.apache.kafka.clients.NodeApiVersions;
import org.apache.kafka.clients.admin.DeleteAclsResult.FilterResults;
import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.internals.AdminMetadataManager;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
@@ -284,6 +285,10 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
/**
* A unit test for KafkaAdminClient.
@@ -7811,6 +7816,37 @@ public class KafkaAdminClientTest {
}
}
+ @Test
+ public void testCallFailWithUnsupportedVersionExceptionDoesNotHaveConcurrentModificationException() throws InterruptedException {
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockCluster(1, 0))) {
+ AdminMetadataManager metadataManager = mock(AdminMetadataManager.class);
+
+ // first false result make sure LeastLoadedBrokerOrActiveKController#provide can go to requestUpdate
+ // second true result make sure LeastLoadedBrokerOrActiveKController#provide can get a node
+ doReturn(false).doReturn(true).when(metadataManager).isReady();
+
+ // make maybeDrainPendingCall throw UnsupportedVersionException and go to Call#fail
+ doThrow(new UnsupportedVersionException("Unsupported version")).doNothing().when(metadataManager).requestUpdate();
+
+ // make sure describeCluster handleUnsupportedVersionException doesn't always return false
+ doReturn(false).when(metadataManager).usingBootstrapControllers();
+ // avoid sending fetchMetadata request
+ doReturn(1L).when(metadataManager).metadataFetchDelayMs(anyLong());
+
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+
+ try (KafkaAdminClient admin = KafkaAdminClient.createInternal(
+ new AdminClientConfig(Collections.emptyMap()), metadataManager, env.kafkaClient(), env.time())) {
+ admin.describeCluster(new DescribeClusterOptions().timeoutMs(1000));
+
+ // make sure maybeDrainPendingCalls doesn't remove duplicate pending calls
+ // the listNodes call will be added again in call.fail and remove one in maybeDrainPendingCalls
+ TestUtils.waitForCondition(() -> env.kafkaClient().inFlightRequestCount() != 0,
+ "Timed out waiting for listNodes request");
+ }
+ }
+ }
+
private static ListClientMetricsResourcesResponse prepareListClientMetricsResourcesResponse(Errors error) {
return new ListClientMetricsResourcesResponse(new ListClientMetricsResourcesResponseData()
.setErrorCode(error.code()));