From 635513c61cbb5b0d8a402081c79445178604cd07 Mon Sep 17 00:00:00 2001 From: huxihx Date: Fri, 6 Sep 2019 10:58:14 +0800 Subject: [PATCH 1/4] KAFKA-8503: Ignore retries config if a custom timeout is provided https://issues.apache.org/jira/browse/KAFKA-8503 The KIP-533 implementation. --- .../kafka/clients/CommonClientConfigs.java | 4 ++ .../kafka/clients/admin/AbstractOptions.java | 4 +- .../clients/admin/AdminClientConfig.java | 14 ++++- .../clients/admin/AlterConfigsOptions.java | 2 +- .../clients/admin/CreateAclsOptions.java | 2 +- .../clients/admin/CreateTopicsOptions.java | 2 +- .../clients/admin/DeleteAclsOptions.java | 2 +- .../clients/admin/DeleteTopicsOptions.java | 2 +- .../clients/admin/DescribeAclsOptions.java | 2 +- .../clients/admin/DescribeClusterOptions.java | 2 +- .../clients/admin/DescribeConfigsOptions.java | 2 +- .../clients/admin/DescribeTopicsOptions.java | 2 +- .../kafka/clients/admin/KafkaAdminClient.java | 47 ++++++++++++-- .../clients/admin/ListTopicsOptions.java | 2 +- .../clients/consumer/ConsumerConfig.java | 5 +- .../clients/admin/KafkaAdminClientTest.java | 62 ++++++++++++++++++- .../kafka/admin/LeaderElectionCommand.scala | 3 +- ...referredReplicaLeaderElectionCommand.scala | 1 + .../api/PlaintextAdminIntegrationTest.scala | 2 +- .../kafka/api/SslAdminIntegrationTest.scala | 2 +- .../admin/LeaderElectionCommandTest.scala | 3 +- .../admin/ReassignPartitionsClusterTest.scala | 1 + 22 files changed, 141 insertions(+), 27 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java index af47e55d75a6a..f1736eebb8da4 100644 --- a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java +++ b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java @@ -141,6 +141,10 @@ public class CommonClientConfigs { + "The value must be set lower than session.timeout.ms, but typically should be set no higher " + "than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances."; + public static final String DEFAULT_API_TIMEOUT_MS_CONFIG = "default.api.timeout.ms"; + public static final String DEFAULT_API_TIMEOUT_MS_DOC = "Specifies the timeout (in milliseconds) for client APIs. " + + "This configuration is used as the default timeout for all client operations that do not explicitly accept a timeout parameter."; + /** * Postprocess the configuration so that exponential backoff is disabled when reconnect backoff * is explicitly configured but the maximum reconnect backoff is not explicitly configured. diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AbstractOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/AbstractOptions.java index ccccf118a1abd..08517e33129a6 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/AbstractOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/AbstractOptions.java @@ -26,7 +26,7 @@ public abstract class AbstractOptions { protected Integer timeoutMs = null; /** - * Set the request timeout in milliseconds for this operation or {@code null} if the default request timeout for the + * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. */ @SuppressWarnings("unchecked") @@ -36,7 +36,7 @@ public T timeoutMs(Integer timeoutMs) { } /** - * The request timeout in milliseconds for this operation or {@code null} if the default request timeout for the + * The api timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. */ public Integer timeoutMs() { diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java index 107eb56d63aed..ad62f1fbcd6d4 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java @@ -30,6 +30,7 @@ import java.util.Set; import static org.apache.kafka.common.config.ConfigDef.Range.atLeast; +import static org.apache.kafka.common.config.ConfigDef.Range.between; import static org.apache.kafka.common.config.ConfigDef.ValidString.in; /** @@ -107,6 +108,7 @@ public class AdminClientConfig extends AbstractConfig { private static final String METRICS_RECORDING_LEVEL_DOC = CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC; public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG; + public static final String DEFAULT_API_TIMEOUT_MS_CONFIG = CommonClientConfigs.DEFAULT_API_TIMEOUT_MS_CONFIG; /** * security.providers @@ -143,7 +145,7 @@ public class AdminClientConfig extends AbstractConfig { RETRY_BACKOFF_MS_DOC) .define(REQUEST_TIMEOUT_MS_CONFIG, Type.INT, - 120000, + 30000, atLeast(0), Importance.MEDIUM, REQUEST_TIMEOUT_MS_DOC) @@ -154,10 +156,16 @@ public class AdminClientConfig extends AbstractConfig { CONNECTIONS_MAX_IDLE_MS_DOC) .define(RETRIES_CONFIG, Type.INT, - 5, - atLeast(0), + Integer.MAX_VALUE, + between(0, Integer.MAX_VALUE), Importance.LOW, CommonClientConfigs.RETRIES_DOC) + .define(DEFAULT_API_TIMEOUT_MS_CONFIG, + Type.INT, + 60000, + atLeast(0), + Importance.MEDIUM, + CommonClientConfigs.DEFAULT_API_TIMEOUT_MS_DOC) .define(METRICS_SAMPLE_WINDOW_MS_CONFIG, Type.LONG, 30000, diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigsOptions.java index 0b280532104e8..ae5dc4c79ac02 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigsOptions.java @@ -32,7 +32,7 @@ public class AlterConfigsOptions extends AbstractOptions { private boolean validateOnly = false; /** - * Set the request timeout in milliseconds for this operation or {@code null} if the default request timeout for the + * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/CreateAclsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/CreateAclsOptions.java index bfb8e32db157f..d5f25fbbfeed2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/CreateAclsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/CreateAclsOptions.java @@ -30,7 +30,7 @@ public class CreateAclsOptions extends AbstractOptions { /** - * Set the request timeout in milliseconds for this operation or {@code null} if the default request timeout for the + * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsOptions.java index a9f1009c2fc7e..88a521fdb6408 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsOptions.java @@ -32,7 +32,7 @@ public class CreateTopicsOptions extends AbstractOptions { private boolean validateOnly = false; /** - * Set the request timeout in milliseconds for this operation or {@code null} if the default request timeout for the + * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteAclsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteAclsOptions.java index 1b67da52f386d..b0a8b4002aa43 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteAclsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteAclsOptions.java @@ -30,7 +30,7 @@ public class DeleteAclsOptions extends AbstractOptions { /** - * Set the request timeout in milliseconds for this operation or {@code null} if the default request timeout for the + * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsOptions.java index 91e38a196fc9a..f9459b7613044 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsOptions.java @@ -30,7 +30,7 @@ public class DeleteTopicsOptions extends AbstractOptions { /** - * Set the request timeout in milliseconds for this operation or {@code null} if the default request timeout for the + * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeAclsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeAclsOptions.java index b17d6a7d0cb98..27d0cb4216bbe 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeAclsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeAclsOptions.java @@ -29,7 +29,7 @@ public class DescribeAclsOptions extends AbstractOptions { /** - * Set the request timeout in milliseconds for this operation or {@code null} if the default request timeout for the + * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeClusterOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeClusterOptions.java index 670feda0d2614..cea2660f2b49a 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeClusterOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeClusterOptions.java @@ -30,7 +30,7 @@ public class DescribeClusterOptions extends AbstractOptions requestBuilder; try { requestBuilder = call.createRequest(timeoutMs); @@ -1002,7 +1040,8 @@ private long sendEligibleCalls(long now) { "Internal error sending %s to %s.", call.callName, node))); continue; } - ClientRequest clientRequest = client.newClientRequest(node.idString(), requestBuilder, now, true); + ClientRequest clientRequest = client.newClientRequest(node.idString(), requestBuilder, now, + true, timeoutMs, null); log.trace("Sending {} to {}. correlationId={}", requestBuilder, node, clientRequest.correlationId()); client.send(clientRequest, now); getOrCreateListValue(callsInFlight, node.idString()).add(call); diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsOptions.java index e288e1828fd79..b0818f6a8e460 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsOptions.java @@ -30,7 +30,7 @@ public class ListTopicsOptions extends AbstractOptions { private boolean listInternal = false; /** - * Set the request timeout in milliseconds for this operation or {@code null} if the default request timeout for the + * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java index 67897cfd54fb3..ce7a595e58613 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java @@ -222,8 +222,7 @@ public class ConsumerConfig extends AbstractConfig { private static final String REQUEST_TIMEOUT_MS_DOC = CommonClientConfigs.REQUEST_TIMEOUT_MS_DOC; /** default.api.timeout.ms */ - public static final String DEFAULT_API_TIMEOUT_MS_CONFIG = "default.api.timeout.ms"; - public static final String DEFAULT_API_TIMEOUT_MS_DOC = "Specifies the timeout (in milliseconds) for consumer APIs that could block. This configuration is used as the default timeout for all consumer operations that do not explicitly accept a timeout parameter."; + public static final String DEFAULT_API_TIMEOUT_MS_CONFIG = CommonClientConfigs.DEFAULT_API_TIMEOUT_MS_CONFIG; /** interceptor.classes */ public static final String INTERCEPTOR_CLASSES_CONFIG = "interceptor.classes"; @@ -447,7 +446,7 @@ public class ConsumerConfig extends AbstractConfig { 60 * 1000, atLeast(0), Importance.MEDIUM, - DEFAULT_API_TIMEOUT_MS_DOC) + CommonClientConfigs.DEFAULT_API_TIMEOUT_MS_DOC) /* default is set to be a bit lower than the server default (10 min), to avoid both client and server closing connection at same time */ .define(CONNECTIONS_MAX_IDLE_MS_CONFIG, Type.LONG, 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 8aa222edce05d..2debb77859a90 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 @@ -38,6 +38,7 @@ import org.apache.kafka.common.acl.AclBindingFilter; import org.apache.kafka.common.acl.AclOperation; import org.apache.kafka.common.acl.AclPermissionType; +import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.config.ConfigResource; import org.apache.kafka.common.errors.AuthenticationException; import org.apache.kafka.common.errors.ClusterAuthorizationException; @@ -178,6 +179,22 @@ public class KafkaAdminClientTest { @Rule final public Timeout globalTimeout = Timeout.millis(120000); + @Test + public void testDefaultApiTimeoutAndRequestTimeoutConflicts() { + AdminClientConfig config = newConfMap(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "500"); + try (Admin client = KafkaAdminClient.createInternal(config, null)) { + fail("Expected KafkaException"); + } catch (KafkaException e) { + assertTrue(e.getCause() instanceof ConfigException); + } + + config = newConfMap(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "3000000"); + try (KafkaAdminClient client = KafkaAdminClient.createInternal(config, null)) { + // default api timeout should be overridden to request timeout. + assertEquals(client.defaultTimeoutMs(), client.requestTimeoutMs()); + } + } + @Test public void testGetOrCreateListValue() { Map> map = new HashMap<>(); @@ -591,7 +608,7 @@ public void testMetadataRetries() throws Exception { try (final AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(Time.SYSTEM, bootstrapCluster, newStrMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999", - AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "10000000", + AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "10000000", AdminClientConfig.RETRIES_CONFIG, "0"))) { // The first request fails with a disconnect @@ -2554,6 +2571,49 @@ public void testGetSubLevelError() { errorsMap, memberIdentities.get(1), "For unit test").getClass()); } + @Test + public void testSingleRequestTimeoutAndRetryWithoutApiTimeout() throws Exception { + HashMap nodes = new HashMap<>(); + MockTime time = new MockTime(); + Node node0 = new Node(0, "localhost", 8121); + nodes.put(0, node0); + Cluster cluster = new Cluster("mockClusterId", nodes.values(), + Arrays.asList(new PartitionInfo("foo", 0, node0, new Node[]{node0}, new Node[]{node0})), + Collections.emptySet(), Collections.emptySet(), + Collections.emptySet(), nodes.get(0)); + + final int requestTimeoutMs = 1000; + final int retryBackoff = 100; + try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time, cluster, + AdminClientConfig.RETRY_BACKOFF_MS_CONFIG, String.valueOf(retryBackoff), + AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, String.valueOf(requestTimeoutMs))) { + env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); + assertEquals(time, env.time()); + assertEquals(env.time(), ((KafkaAdminClient) env.adminClient()).time()); + + final int apiTimeoutMs = 3000; + final long startTimeMs = time.milliseconds(); + final ListTopicsResult result = env.adminClient().listTopics(new ListTopicsOptions().timeoutMs(apiTimeoutMs)); + + // Wait until the first attempt has failed, then advance the time + TestUtils.waitForCondition(() -> env.kafkaClient().hasInFlightRequests(), "Timed out waiting for inFlightRequests"); + time.sleep(requestTimeoutMs + retryBackoff); + final long betweenTimeoutMs = time.milliseconds(); + + // Since api timeout bound is not hit, AdminClient should add the retry call to the queue + TestUtils.waitForCondition(() -> ((KafkaAdminClient) env.adminClient()).numPendingCalls() == 1, + "Failed to add retry listTopics call"); + env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE)); + time.sleep(requestTimeoutMs); + + assertEquals(apiTimeoutMs - requestTimeoutMs - retryBackoff, + KafkaAdminClient.calcTimeoutMsRemainingAsInt(betweenTimeoutMs, apiTimeoutMs + startTimeMs)); + assertEquals(result.listings().get().size(), 1); + assertEquals("foo", result.listings().get().iterator().next().name()); + + } + } + private static MemberDescription convertToMemberDescriptions(DescribedGroupMember member, MemberAssignment assignment) { return new MemberDescription(member.memberId(), diff --git a/core/src/main/scala/kafka/admin/LeaderElectionCommand.scala b/core/src/main/scala/kafka/admin/LeaderElectionCommand.scala index 3e30a474f2855..e71702ab582b3 100644 --- a/core/src/main/scala/kafka/admin/LeaderElectionCommand.scala +++ b/core/src/main/scala/kafka/admin/LeaderElectionCommand.scala @@ -79,7 +79,8 @@ object LeaderElectionCommand extends Logging { AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, commandOptions.options.valueOf(commandOptions.bootstrapServer) ) - props.setProperty(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, timeout.toMillis.toString) + props.setProperty(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, timeout.toMillis.toString) + props.setProperty(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, (timeout.toMillis / 2).toString) Admin.create(props) } diff --git a/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala b/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala index 2c0eea050be80..5ddf047b3521a 100755 --- a/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala +++ b/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala @@ -74,6 +74,7 @@ object PreferredReplicaLeaderElectionCommand extends Logging { new Properties() adminProps.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, commandOpts.options.valueOf(commandOpts.bootstrapServerOpt)) adminProps.setProperty(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, timeout.toString) + adminProps.setProperty(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, (timeout * 2).toString) new AdminClientCommand(adminProps) } diff --git a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala index ca3c5a8c107c2..c33df2dafb650 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala @@ -986,7 +986,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest { @Test def testCallInFlightTimeouts(): Unit = { val config = createConfig() - config.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "100000000") + config.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "100000000") val factory = new KafkaAdminClientTest.FailureInjectingTimeoutProcessorFactory() client = KafkaAdminClientTest.createInternal(new AdminClientConfig(config), factory) val future = client.createTopics(Seq("mytopic", "mytopic2").map(new NewTopic(_, 1, 1.toShort)).asJava, diff --git a/core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala index 5f987b805cccd..142dbca5942ec 100644 --- a/core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala @@ -238,7 +238,7 @@ class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest { private def createAdminClient: Admin = { val config = createConfig() - config.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "40000") + config.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "40000") val client = Admin.create(config) adminClients += client client diff --git a/core/src/test/scala/unit/kafka/admin/LeaderElectionCommandTest.scala b/core/src/test/scala/unit/kafka/admin/LeaderElectionCommandTest.scala index e328f2afb206a..1b8069ac26da0 100644 --- a/core/src/test/scala/unit/kafka/admin/LeaderElectionCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/LeaderElectionCommandTest.scala @@ -309,7 +309,8 @@ object LeaderElectionCommandTest { def createConfig(servers: Seq[KafkaServer]): Map[String, Object] = { Map( AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG -> bootstrapServers(servers), - AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG -> "20000" + AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG -> "20000", + AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG -> "10000" ) } diff --git a/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala b/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala index 1108d999bd744..1cbf33e0f0b81 100644 --- a/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala @@ -67,6 +67,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging { val props = new Properties() props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, TestUtils.getBrokerListStrFromServers(servers)) props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "10000") + props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "15000") Admin.create(props) } From edc0711cdb10fa4aea8c4bf67fbb8bca82ed22e4 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Mon, 30 Dec 2019 11:28:41 -0800 Subject: [PATCH 2/4] Add test cases and minor cleanups --- .../kafka/clients/admin/AbstractOptions.java | 4 +- .../clients/admin/AlterConfigsOptions.java | 2 +- .../clients/admin/CreateAclsOptions.java | 2 +- .../clients/admin/CreateTopicsOptions.java | 2 +- .../clients/admin/DeleteAclsOptions.java | 2 +- .../clients/admin/DeleteTopicsOptions.java | 2 +- .../clients/admin/DescribeAclsOptions.java | 2 +- .../clients/admin/DescribeClusterOptions.java | 2 +- .../clients/admin/DescribeConfigsOptions.java | 2 +- .../clients/admin/DescribeTopicsOptions.java | 2 +- .../kafka/clients/admin/KafkaAdminClient.java | 37 ++-- .../clients/admin/ListTopicsOptions.java | 2 +- .../clients/admin/KafkaAdminClientTest.java | 208 +++++++++++------- 13 files changed, 149 insertions(+), 120 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AbstractOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/AbstractOptions.java index 08517e33129a6..2312fe4b81dd2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/AbstractOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/AbstractOptions.java @@ -26,7 +26,7 @@ public abstract class AbstractOptions { protected Integer timeoutMs = null; /** - * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the + * Set the timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. */ @SuppressWarnings("unchecked") @@ -36,7 +36,7 @@ public T timeoutMs(Integer timeoutMs) { } /** - * The api timeout in milliseconds for this operation or {@code null} if the default api timeout for the + * The timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. */ public Integer timeoutMs() { diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigsOptions.java index ae5dc4c79ac02..fc933c4b1dba0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigsOptions.java @@ -32,7 +32,7 @@ public class AlterConfigsOptions extends AbstractOptions { private boolean validateOnly = false; /** - * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the + * Set the timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/CreateAclsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/CreateAclsOptions.java index d5f25fbbfeed2..ad4ae74ff26ec 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/CreateAclsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/CreateAclsOptions.java @@ -30,7 +30,7 @@ public class CreateAclsOptions extends AbstractOptions { /** - * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the + * Set the timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsOptions.java index 88a521fdb6408..cd03bc62e24a3 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsOptions.java @@ -32,7 +32,7 @@ public class CreateTopicsOptions extends AbstractOptions { private boolean validateOnly = false; /** - * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the + * Set the timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteAclsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteAclsOptions.java index b0a8b4002aa43..7c250e10b3e27 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteAclsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteAclsOptions.java @@ -30,7 +30,7 @@ public class DeleteAclsOptions extends AbstractOptions { /** - * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the + * Set the timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsOptions.java index f9459b7613044..def02867b672c 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsOptions.java @@ -30,7 +30,7 @@ public class DeleteTopicsOptions extends AbstractOptions { /** - * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the + * Set the timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeAclsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeAclsOptions.java index 27d0cb4216bbe..e44d58473df92 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeAclsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeAclsOptions.java @@ -29,7 +29,7 @@ public class DescribeAclsOptions extends AbstractOptions { /** - * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the + * Set the timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeClusterOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeClusterOptions.java index cea2660f2b49a..2eac1f055f6e0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeClusterOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeClusterOptions.java @@ -30,7 +30,7 @@ public class DescribeClusterOptions extends AbstractOptions requestBuilder; try { - requestBuilder = call.createRequest(timeoutMs); + requestBuilder = call.createRequest(requestTimeoutMs); } catch (Throwable throwable) { call.fail(now, new KafkaException(String.format( "Internal error sending %s to %s.", call.callName, node))); continue; } ClientRequest clientRequest = client.newClientRequest(node.idString(), requestBuilder, now, - true, timeoutMs, null); + true, requestTimeoutMs, null); log.trace("Sending {} to {}. correlationId={}", requestBuilder, node, clientRequest.correlationId()); client.send(clientRequest, now); getOrCreateListValue(callsInFlight, node.idString()).add(call); @@ -1339,7 +1328,7 @@ void call(Call call, long now) { * Create a new metadata call. */ private Call makeMetadataCall(long now) { - return new Call(true, "fetchMetadata", calcDeadlineMs(now, defaultTimeoutMs), + return new Call(true, "fetchMetadata", calcDeadlineMs(now, requestTimeoutMs), new MetadataUpdateNodeIdProvider()) { @Override public MetadataRequest.Builder createRequest(int timeoutMs) { diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsOptions.java index b0818f6a8e460..5a8f2b4ad1514 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsOptions.java @@ -30,7 +30,7 @@ public class ListTopicsOptions extends AbstractOptions { private boolean listInternal = false; /** - * Set the api timeout in milliseconds for this operation or {@code null} if the default api timeout for the + * Set the timeout in milliseconds for this operation or {@code null} if the default api timeout for the * AdminClient should be used. * */ 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 2debb77859a90..6a15269971ff3 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 @@ -109,11 +109,11 @@ import org.apache.kafka.common.requests.ListGroupsResponse; import org.apache.kafka.common.requests.ListOffsetResponse; import org.apache.kafka.common.requests.ListOffsetResponse.PartitionData; -import org.apache.kafka.common.requests.MetadataResponse.PartitionMetadata; -import org.apache.kafka.common.requests.MetadataResponse.TopicMetadata; import org.apache.kafka.common.requests.ListPartitionReassignmentsResponse; import org.apache.kafka.common.requests.MetadataRequest; import org.apache.kafka.common.requests.MetadataResponse; +import org.apache.kafka.common.requests.MetadataResponse.PartitionMetadata; +import org.apache.kafka.common.requests.MetadataResponse.TopicMetadata; import org.apache.kafka.common.requests.OffsetCommitResponse; import org.apache.kafka.common.requests.OffsetDeleteResponse; import org.apache.kafka.common.requests.OffsetFetchResponse; @@ -124,9 +124,7 @@ import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; -import org.apache.kafka.test.TestCondition; import org.apache.kafka.test.TestUtils; -import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; import org.junit.rules.Timeout; @@ -145,6 +143,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.OptionalInt; import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; @@ -181,18 +180,10 @@ public class KafkaAdminClientTest { @Test public void testDefaultApiTimeoutAndRequestTimeoutConflicts() { - AdminClientConfig config = newConfMap(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "500"); - try (Admin client = KafkaAdminClient.createInternal(config, null)) { - fail("Expected KafkaException"); - } catch (KafkaException e) { - assertTrue(e.getCause() instanceof ConfigException); - } - - config = newConfMap(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "3000000"); - try (KafkaAdminClient client = KafkaAdminClient.createInternal(config, null)) { - // default api timeout should be overridden to request timeout. - assertEquals(client.defaultTimeoutMs(), client.requestTimeoutMs()); - } + final AdminClientConfig config = newConfMap(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "500"); + KafkaException exception = assertThrows(KafkaException.class, + () -> KafkaAdminClient.createInternal(config, null)); + assertTrue(exception.getCause() instanceof ConfigException); } @Test @@ -865,54 +856,6 @@ public void testElectLeaders() throws Exception { } } - /** - * Test handling timeouts. - */ - @Ignore // The test is flaky. Should be renabled when this JIRA is fixed: https://issues.apache.org/jira/browse/KAFKA-5792 - @Test - public void testHandleTimeout() throws Exception { - MockTime time = new MockTime(); - try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time, - mockCluster(1, 0), - AdminClientConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG, "1", - AdminClientConfig.RECONNECT_BACKOFF_MS_CONFIG, "1")) { - env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); - assertEquals(time, env.time()); - assertEquals(env.time(), ((KafkaAdminClient) env.adminClient()).time()); - - // Make a request with an extremely short timeout. - // Then wait for it to fail by not supplying any response. - log.info("Starting AdminClient#listTopics..."); - final ListTopicsResult result = env.adminClient().listTopics(new ListTopicsOptions().timeoutMs(1000)); - TestUtils.waitForCondition(new TestCondition() { - @Override - public boolean conditionMet() { - return env.kafkaClient().hasInFlightRequests(); - } - }, "Timed out waiting for inFlightRequests"); - time.sleep(5000); - TestUtils.waitForCondition(new TestCondition() { - @Override - public boolean conditionMet() { - return result.listings().isDone(); - } - }, "Timed out waiting for listTopics to complete"); - TestUtils.assertFutureError(result.listings(), TimeoutException.class); - log.info("Verified the error result of AdminClient#listTopics"); - - // The next request should succeed. - time.sleep(5000); - env.kafkaClient().prepareResponse(new DescribeConfigsResponse(0, - Collections.singletonMap(new ConfigResource(ConfigResource.Type.TOPIC, "foo"), - new DescribeConfigsResponse.Config(ApiError.NONE, - Collections.emptySet())))); - DescribeConfigsResult result2 = env.adminClient().describeConfigs(Collections.singleton( - new ConfigResource(ConfigResource.Type.TOPIC, "foo"))); - time.sleep(5000); - result2.values().get(new ConfigResource(ConfigResource.Type.TOPIC, "foo")).get(); - } - } - @Test public void testDescribeConfigs() throws Exception { try (AdminClientUnitTestEnv env = mockClientEnv()) { @@ -2572,7 +2515,7 @@ public void testGetSubLevelError() { } @Test - public void testSingleRequestTimeoutAndRetryWithoutApiTimeout() throws Exception { + public void testSuccessfulRetryAfterRequestTimeout() throws Exception { HashMap nodes = new HashMap<>(); MockTime time = new MockTime(); Node node0 = new Node(0, "localhost", 8121); @@ -2583,34 +2526,131 @@ public void testSingleRequestTimeoutAndRetryWithoutApiTimeout() throws Exception Collections.emptySet(), nodes.get(0)); final int requestTimeoutMs = 1000; - final int retryBackoff = 100; + final int retryBackoffMs = 100; + final int apiTimeoutMs = 3000; + try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time, cluster, - AdminClientConfig.RETRY_BACKOFF_MS_CONFIG, String.valueOf(retryBackoff), + AdminClientConfig.RETRY_BACKOFF_MS_CONFIG, String.valueOf(retryBackoffMs), AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, String.valueOf(requestTimeoutMs))) { env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); - assertEquals(time, env.time()); - assertEquals(env.time(), ((KafkaAdminClient) env.adminClient()).time()); - final int apiTimeoutMs = 3000; - final long startTimeMs = time.milliseconds(); - final ListTopicsResult result = env.adminClient().listTopics(new ListTopicsOptions().timeoutMs(apiTimeoutMs)); + final ListTopicsResult result = env.adminClient() + .listTopics(new ListTopicsOptions().timeoutMs(apiTimeoutMs)); - // Wait until the first attempt has failed, then advance the time - TestUtils.waitForCondition(() -> env.kafkaClient().hasInFlightRequests(), "Timed out waiting for inFlightRequests"); - time.sleep(requestTimeoutMs + retryBackoff); - final long betweenTimeoutMs = time.milliseconds(); + // Wait until the first attempt has been sent, then advance the time + TestUtils.waitForCondition(() -> env.kafkaClient().hasInFlightRequests(), + "Timed out waiting for Metadata request to be sent"); + time.sleep(requestTimeoutMs + 1); - // Since api timeout bound is not hit, AdminClient should add the retry call to the queue - TestUtils.waitForCondition(() -> ((KafkaAdminClient) env.adminClient()).numPendingCalls() == 1, - "Failed to add retry listTopics call"); - env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE)); - time.sleep(requestTimeoutMs); + // Wait for the request to be timed out before backing off + TestUtils.waitForCondition(() -> !env.kafkaClient().hasInFlightRequests(), + "Timed out waiting for inFlightRequests to be timed out"); + time.sleep(retryBackoffMs); - assertEquals(apiTimeoutMs - requestTimeoutMs - retryBackoff, - KafkaAdminClient.calcTimeoutMsRemainingAsInt(betweenTimeoutMs, apiTimeoutMs + startTimeMs)); - assertEquals(result.listings().get().size(), 1); + // Since api timeout bound is not hit, AdminClient should retry + TestUtils.waitForCondition(() -> env.kafkaClient().hasInFlightRequests(), + "Failed to retry Metadata request"); + env.kafkaClient().respond(prepareMetadataResponse(cluster, Errors.NONE)); + + assertEquals(1, result.listings().get().size()); assertEquals("foo", result.listings().get().iterator().next().name()); + } + } + + @Test + public void testDefaultApiTimeout() throws Exception { + testApiTimeout(1500, 3000, OptionalInt.empty()); + } + + @Test + public void testDefaultApiTimeoutOverride() throws Exception { + testApiTimeout(1500, 10000, OptionalInt.of(3000)); + } + + private void testApiTimeout(int requestTimeoutMs, + int defaultApiTimeoutMs, + OptionalInt overrideApiTimeoutMs) throws Exception { + HashMap nodes = new HashMap<>(); + MockTime time = new MockTime(); + Node node0 = new Node(0, "localhost", 8121); + nodes.put(0, node0); + Cluster cluster = new Cluster("mockClusterId", nodes.values(), + Arrays.asList(new PartitionInfo("foo", 0, node0, new Node[]{node0}, new Node[]{node0})), + Collections.emptySet(), Collections.emptySet(), + Collections.emptySet(), nodes.get(0)); + + final int retryBackoffMs = 100; + final int effectiveTimeoutMs = overrideApiTimeoutMs.orElse(defaultApiTimeoutMs); + assertEquals("This test expects the effective timeout to be twice the request timeout", + 2 * requestTimeoutMs, effectiveTimeoutMs); + + try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time, cluster, + AdminClientConfig.RETRY_BACKOFF_MS_CONFIG, String.valueOf(retryBackoffMs), + AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, String.valueOf(requestTimeoutMs), + AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, String.valueOf(defaultApiTimeoutMs))) { + env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); + + ListTopicsOptions options = new ListTopicsOptions(); + overrideApiTimeoutMs.ifPresent(options::timeoutMs); + + final ListTopicsResult result = env.adminClient().listTopics(options); + + // Wait until the first attempt has been sent, then advance the time + TestUtils.waitForCondition(() -> env.kafkaClient().hasInFlightRequests(), + "Timed out waiting for Metadata request to be sent"); + time.sleep(requestTimeoutMs + 1); + + // Wait for the request to be timed out before backing off + TestUtils.waitForCondition(() -> !env.kafkaClient().hasInFlightRequests(), + "Timed out waiting for inFlightRequests to be timed out"); + time.sleep(retryBackoffMs); + + // Since api timeout bound is not hit, AdminClient should retry + TestUtils.waitForCondition(() -> env.kafkaClient().hasInFlightRequests(), + "Timed out waiting for Metadata request to be sent"); + time.sleep(requestTimeoutMs + 1); + + TestUtils.assertFutureThrows(result.future, TimeoutException.class); + } + } + + @Test + public void testRequestTimeoutExceedingDefaultApiTimeout() throws Exception { + HashMap nodes = new HashMap<>(); + MockTime time = new MockTime(); + Node node0 = new Node(0, "localhost", 8121); + nodes.put(0, node0); + Cluster cluster = new Cluster("mockClusterId", nodes.values(), + Arrays.asList(new PartitionInfo("foo", 0, node0, new Node[]{node0}, new Node[]{node0})), + Collections.emptySet(), Collections.emptySet(), + Collections.emptySet(), nodes.get(0)); + + // This test assumes the default api timeout value of 60000. When the request timeout + // is set to something larger, we should adjust the api timeout accordingly for compatibility. + + final int retryBackoffMs = 100; + final int requestTimeoutMs = 120000; + + try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time, cluster, + AdminClientConfig.RETRY_BACKOFF_MS_CONFIG, String.valueOf(retryBackoffMs), + AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, String.valueOf(requestTimeoutMs))) { + env.kafkaClient().setNodeApiVersions(NodeApiVersions.create()); + + ListTopicsOptions options = new ListTopicsOptions(); + + final ListTopicsResult result = env.adminClient().listTopics(options); + + // Wait until the first attempt has been sent, then advance the time by the default api timeout + TestUtils.waitForCondition(() -> env.kafkaClient().hasInFlightRequests(), + "Timed out waiting for Metadata request to be sent"); + time.sleep(60001); + + // The in-flight request should not be cancelled + assertTrue(env.kafkaClient().hasInFlightRequests()); + // Now sleep the remaining time for the request timeout to expire + time.sleep(60000); + TestUtils.assertFutureThrows(result.future, TimeoutException.class); } } From 4ad7c907b5e4b226d3f5cf27181c1251a8a5067f Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Wed, 29 Jan 2020 16:30:44 -0800 Subject: [PATCH 3/4] Make request timeout and default api timeout final fields --- .../kafka/clients/admin/KafkaAdminClient.java | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 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 9531bb03c723f..420786c670e79 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 @@ -264,12 +264,12 @@ public class KafkaAdminClient extends AdminClient { /** * The default timeout to use for an operation. */ - private int defaultApiTimeoutMs; + private final int defaultApiTimeoutMs; /** * The timeout to use for a single request. */ - private int requestTimeoutMs; + private final int requestTimeoutMs; /** * The name of this AdminClient instance. @@ -508,7 +508,8 @@ private KafkaAdminClient(AdminClientConfig config, LogContext logContext) { this.clientId = clientId; this.log = logContext.logger(KafkaAdminClient.class); - configureDefaultApiTimeoutMsAndRequestTimeoutMs(config); + this.requestTimeoutMs = config.getInt(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG); + this.defaultApiTimeoutMs = configureDefaultApiTimeoutMs(config); this.time = time; this.metadataManager = metadataManager; this.metrics = metrics; @@ -533,21 +534,22 @@ private KafkaAdminClient(AdminClientConfig config, * * @param config The configuration */ - private void configureDefaultApiTimeoutMsAndRequestTimeoutMs(AdminClientConfig config) { - this.requestTimeoutMs = config.getInt(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG); - this.defaultApiTimeoutMs = config.getInt(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG); + private int configureDefaultApiTimeoutMs(AdminClientConfig config) { + int requestTimeoutMs = config.getInt(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG); + int defaultApiTimeoutMs = config.getInt(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG); - if (this.defaultApiTimeoutMs < this.requestTimeoutMs) { + if (defaultApiTimeoutMs < requestTimeoutMs) { if (config.originals().containsKey(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG)) { throw new ConfigException("The specified value of " + AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG + " must be no smaller than the value of " + AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG + "."); } else { log.warn("Overriding the default value for {} ({}) with the explicitly configured request timeout {}", AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, this.defaultApiTimeoutMs, - this.requestTimeoutMs); - this.defaultApiTimeoutMs = this.requestTimeoutMs; + requestTimeoutMs); + return requestTimeoutMs; } } + return defaultApiTimeoutMs; } @Override From c83cf9d4c951cbf57ece9e9903338e34007c7473 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Thu, 30 Jan 2020 08:50:52 -0800 Subject: [PATCH 4/4] Fix description of config --- .../main/java/org/apache/kafka/clients/CommonClientConfigs.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java index f1736eebb8da4..d18c0ed9e89f8 100644 --- a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java +++ b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java @@ -143,7 +143,7 @@ public class CommonClientConfigs { public static final String DEFAULT_API_TIMEOUT_MS_CONFIG = "default.api.timeout.ms"; public static final String DEFAULT_API_TIMEOUT_MS_DOC = "Specifies the timeout (in milliseconds) for client APIs. " + - "This configuration is used as the default timeout for all client operations that do not explicitly accept a timeout parameter."; + "This configuration is used as the default timeout for all client operations that do not specify a timeout parameter."; /** * Postprocess the configuration so that exponential backoff is disabled when reconnect backoff