You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ASF GitHub Bot (JIRA)" <ji...@apache.org> on 2018/07/09 06:56:00 UTC
[jira] [Commented] (KAFKA-7033) Modify AbstractOptions's timeoutMs
as Long type
[ https://issues.apache.org/jira/browse/KAFKA-7033?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16536588#comment-16536588 ]
ASF GitHub Bot commented on KAFKA-7033:
---------------------------------------
darionyaphet closed pull request #5186: [KAFKA-7033] Modify AbstractOptions's timeoutMs as Long type
URL: https://github.com/apache/kafka/pull/5186
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
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 ccccf118a1a..5925e19272e 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
@@ -23,14 +23,14 @@
*/
public abstract class AbstractOptions<T extends AbstractOptions> {
- protected Integer timeoutMs = null;
+ protected Long timeoutMs = null;
/**
* Set the request timeout in milliseconds for this operation or {@code null} if the default request timeout for the
* AdminClient should be used.
*/
@SuppressWarnings("unchecked")
- public T timeoutMs(Integer timeoutMs) {
+ public T timeoutMs(Long timeoutMs) {
this.timeoutMs = timeoutMs;
return (T) this;
}
@@ -39,7 +39,7 @@ public T timeoutMs(Integer timeoutMs) {
* The request timeout in milliseconds for this operation or {@code null} if the default request timeout for the
* AdminClient should be used.
*/
- public Integer timeoutMs() {
+ public Long timeoutMs() {
return 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 2dbeba2c937..0fb80407976 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
@@ -37,7 +37,7 @@
*
*/
// This method is retained to keep binary compatibility with 0.11
- public AlterConfigsOptions timeoutMs(Integer timeoutMs) {
+ public AlterConfigsOptions timeoutMs(Long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
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 a7b92ba93d9..c90c26bcd4b 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
@@ -35,7 +35,7 @@
*
*/
// This method is retained to keep binary compatibility with 0.11
- public CreateAclsOptions timeoutMs(Integer timeoutMs) {
+ public CreateAclsOptions timeoutMs(Long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
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 7cd0df86e8b..07aceb59d4a 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
@@ -37,7 +37,7 @@
*
*/
// This method is retained to keep binary compatibility with 0.11
- public CreateTopicsOptions timeoutMs(Integer timeoutMs) {
+ public CreateTopicsOptions timeoutMs(Long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
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 56f4b78b64c..07dcbd53736 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
@@ -35,7 +35,7 @@
*
*/
// This method is retained to keep binary compatibility with 0.11
- public DeleteAclsOptions timeoutMs(Integer timeoutMs) {
+ public DeleteAclsOptions timeoutMs(Long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
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 27c9af9f17f..f4b93d11eb7 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
@@ -35,7 +35,7 @@
*
*/
// This method is retained to keep binary compatibility with 0.11
- public DeleteTopicsOptions timeoutMs(Integer timeoutMs) {
+ public DeleteTopicsOptions timeoutMs(Long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
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 89ba4bc9bec..5f7264f43ca 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
@@ -34,7 +34,7 @@
*
*/
// This method is retained to keep binary compatibility with 0.11
- public DescribeAclsOptions timeoutMs(Integer timeoutMs) {
+ public DescribeAclsOptions timeoutMs(Long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
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 92640fd54ea..ebeef5cf820 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
@@ -33,7 +33,7 @@
*
*/
// This method is retained to keep binary compatibility with 0.11
- public DescribeClusterOptions timeoutMs(Integer timeoutMs) {
+ public DescribeClusterOptions timeoutMs(Long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeConfigsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeConfigsOptions.java
index aa667af3df2..2e2c419bf57 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeConfigsOptions.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeConfigsOptions.java
@@ -37,7 +37,7 @@
*
*/
// This method is retained to keep binary compatibility with 0.11
- public DescribeConfigsOptions timeoutMs(Integer timeoutMs) {
+ public DescribeConfigsOptions timeoutMs(Long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsOptions.java
index cc3d420e134..f47423dfe9f 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsOptions.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsOptions.java
@@ -35,7 +35,7 @@
*
*/
// This method is retained to keep binary compatibility with 0.11
- public DescribeTopicsOptions timeoutMs(Integer timeoutMs) {
+ public DescribeTopicsOptions timeoutMs(Long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
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 450de06fcd1..7288f7d02ef 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
@@ -171,7 +171,7 @@
/**
* The default timeout to use for an operation.
*/
- private final int defaultTimeoutMs;
+ private final long defaultTimeoutMs;
/**
* The name of this AdminClient instance.
@@ -292,7 +292,7 @@ static String generateClientId(AdminClientConfig config) {
*
* @return The deadline in milliseconds.
*/
- private long calcDeadlineMs(long now, Integer optionTimeoutMs) {
+ private long calcDeadlineMs(long now, Long optionTimeoutMs) {
if (optionTimeoutMs != null)
return now + Math.max(0, optionTimeoutMs);
return now + defaultTimeoutMs;
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 14314946468..875a52d38be 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
@@ -35,7 +35,7 @@
*
*/
// This method is retained to keep binary compatibility with 0.11
- public ListTopicsOptions timeoutMs(Integer timeoutMs) {
+ public ListTopicsOptions timeoutMs(Long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
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 5cb6bbc9761..d353df23ece 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
@@ -228,7 +228,7 @@ public void testTimeoutWithoutMetadata() throws Exception {
env.kafkaClient().prepareResponse(new CreateTopicsResponse(Collections.singletonMap("myTopic", new ApiError(Errors.NONE, ""))));
KafkaFuture<Void> future = env.adminClient().createTopics(
Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(0, asList(0, 1, 2)))),
- new CreateTopicsOptions().timeoutMs(1000)).all();
+ new CreateTopicsOptions().timeoutMs(1000L)).all();
assertFutureError(future, TimeoutException.class);
}
}
@@ -254,7 +254,7 @@ public void testConnectionFailureOnMetadataUpdate() throws Exception {
KafkaFuture<Void> future = env.adminClient().createTopics(
Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(0, asList(0, 1, 2)))),
- new CreateTopicsOptions().timeoutMs(10000)).all();
+ new CreateTopicsOptions().timeoutMs(10000L)).all();
future.get();
}
@@ -281,7 +281,7 @@ public void testUnreachableBootstrapServer() throws Exception {
KafkaFuture<Void> future = env.adminClient().createTopics(
Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(0, asList(0, 1, 2)))),
- new CreateTopicsOptions().timeoutMs(10000)).all();
+ new CreateTopicsOptions().timeoutMs(10000L)).all();
future.get();
}
@@ -306,7 +306,7 @@ public void testPropagatedMetadataFetchException() throws Exception {
env.kafkaClient().prepareResponse(new CreateTopicsResponse(Collections.singletonMap("myTopic", new ApiError(Errors.NONE, ""))));
KafkaFuture<Void> future = env.adminClient().createTopics(
Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(0, asList(0, 1, 2)))),
- new CreateTopicsOptions().timeoutMs(1000)).all();
+ new CreateTopicsOptions().timeoutMs(1000L)).all();
assertFutureError(future, SaslAuthenticationException.class);
}
}
@@ -320,7 +320,7 @@ public void testCreateTopics() throws Exception {
new CreateTopicsResponse(Collections.singletonMap("myTopic", new ApiError(Errors.NONE, ""))));
KafkaFuture<Void> future = env.adminClient().createTopics(
Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(0, asList(0, 1, 2)))),
- new CreateTopicsOptions().timeoutMs(10000)).all();
+ new CreateTopicsOptions().timeoutMs(10000L)).all();
future.get();
}
}
@@ -356,7 +356,7 @@ public void testCreateTopicsRetryBackoff() throws Exception {
KafkaFuture<Void> future = env.adminClient().createTopics(
Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(0, asList(0, 1, 2)))),
- new CreateTopicsOptions().timeoutMs(10000)).all();
+ new CreateTopicsOptions().timeoutMs(10000L)).all();
// Wait until the first attempt has failed, then advance the time
TestUtils.waitForCondition(() -> mockClient.numAwaitingResponses() == 1,
@@ -388,7 +388,7 @@ public void testCreateTopicsHandleNotControllerException() throws Exception {
env.cluster().nodeById(1));
KafkaFuture<Void> future = env.adminClient().createTopics(
Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(0, asList(0, 1, 2)))),
- new CreateTopicsOptions().timeoutMs(10000)).all();
+ new CreateTopicsOptions().timeoutMs(10000L)).all();
future.get();
}
}
@@ -486,7 +486,7 @@ private void callAdminClientApisAndExpectAnAuthenticationError(AdminClientUnitTe
try {
env.adminClient().createTopics(
Collections.singleton(new NewTopic("myTopic", Collections.singletonMap(0, asList(0, 1, 2)))),
- new CreateTopicsOptions().timeoutMs(10000)).all().get();
+ new CreateTopicsOptions().timeoutMs(10000L)).all().get();
fail("Expected an authentication error.");
} catch (ExecutionException e) {
assertTrue("Expected an authentication error, but got " + Utils.stackTrace(e),
@@ -660,7 +660,7 @@ public void testHandleTimeout() throws Exception {
// 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));
+ final ListTopicsResult result = env.adminClient().listTopics(new ListTopicsOptions().timeoutMs(1000L));
TestUtils.waitForCondition(new TestCondition() {
@Override
public boolean conditionMet() {
diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index c1b384fd1fa..495035c897a 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -252,7 +252,7 @@ object ConfigCommand extends Config {
throw new InvalidConfigException(s"All sensitive broker config entries must be specified for --alter, missing entries: ${sensitiveEntries.keySet}")
val newConfig = new JConfig(newEntries.asJava.values)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000L).validateOnly(false)
adminClient.alterConfigs(Map(configResource -> newConfig).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
if (entityName.nonEmpty)
diff --git a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
index 4d9da90bc69..14029830cb1 100755
--- a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
+++ b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
@@ -602,7 +602,7 @@ class ReassignPartitionsCommand(zkClient: KafkaZkClient,
private def alterReplicaLogDirsIgnoreReplicaNotAvailable(replicaAssignment: Map[TopicPartitionReplica, String],
adminClient: JAdminClient,
timeoutMs: Long): Set[TopicPartitionReplica] = {
- val alterReplicaLogDirsResult = adminClient.alterReplicaLogDirs(replicaAssignment.asJava, new AlterReplicaLogDirsOptions().timeoutMs(timeoutMs.toInt))
+ val alterReplicaLogDirsResult = adminClient.alterReplicaLogDirs(replicaAssignment.asJava, new AlterReplicaLogDirsOptions().timeoutMs(timeoutMs.toLong))
val replicasAssignedToFutureDir = alterReplicaLogDirsResult.values().asScala.flatMap { case (replica, future) => {
try {
future.get()
diff --git a/core/src/main/scala/kafka/tools/StreamsResetter.java b/core/src/main/scala/kafka/tools/StreamsResetter.java
index 3c045c69eb2..b9ff0721022 100644
--- a/core/src/main/scala/kafka/tools/StreamsResetter.java
+++ b/core/src/main/scala/kafka/tools/StreamsResetter.java
@@ -156,7 +156,7 @@ public int run(final String[] args,
private void validateNoActiveConsumers(final String groupId,
final AdminClient adminClient) throws ExecutionException, InterruptedException {
final DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(Arrays.asList(groupId),
- (new DescribeConsumerGroupsOptions()).timeoutMs(10 * 1000));
+ (new DescribeConsumerGroupsOptions()).timeoutMs(10 * 1000L));
final List<MemberDescription> members =
new ArrayList<MemberDescription>(describeResult.describedGroups().get(groupId).get().members());
if (!members.isEmpty()) {
diff --git a/tools/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java b/tools/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java
index 3d4871a1729..eb73ee6ba1c 100644
--- a/tools/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java
+++ b/tools/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java
@@ -97,8 +97,8 @@ public static void addConfigsToProperties(
}
}
- private static final int ADMIN_REQUEST_TIMEOUT = 25000;
- private static final int CREATE_TOPICS_CALL_TIMEOUT = 180000;
+ private static final long ADMIN_REQUEST_TIMEOUT = 25000;
+ private static final long CREATE_TOPICS_CALL_TIMEOUT = 180000;
private static final int MAX_CREATE_TOPICS_BATCH_SIZE = 10;
//Map<String, Map<Integer, List<Integer>>> topics) throws Throwable {
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
> Modify AbstractOptions's timeoutMs as Long type
> -----------------------------------------------
>
> Key: KAFKA-7033
> URL: https://issues.apache.org/jira/browse/KAFKA-7033
> Project: Kafka
> Issue Type: Improvement
> Components: clients
> Affects Versions: 1.1.0
> Reporter: darion yaphet
> Priority: Minor
>
> Currently AbstractOptions's timeoutMs is Integer and using Long to represent timeout Millisecond maybe better .
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)