diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java index e7514f815f9db..2357ba8fff6bf 100644 --- a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java +++ b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java @@ -19,6 +19,7 @@ import java.util.Map; import java.util.concurrent.atomic.AtomicReference; +import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.network.ChannelBuilders; import org.apache.kafka.common.protocol.SecurityProtocol; import org.apache.kafka.common.network.ChannelBuilder; @@ -79,4 +80,10 @@ public static ChannelBuilder createChannelBuilder(Map configs) { return ChannelBuilders.create(securityProtocol, SSLFactory.Mode.CLIENT, configs); } + public static long maybeGetRemainingTime(long start, long now, long timeout, String message) { + long elapsed = now - start; + if (elapsed > timeout) + throw new TimeoutException(message); + else return timeout - elapsed; + } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/CommittedTimeoutKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/CommittedTimeoutKafkaConsumer.java new file mode 100644 index 0000000000000..f15f127fe26ae --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/CommittedTimeoutKafkaConsumer.java @@ -0,0 +1,36 @@ +/** + * 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.clients.consumer; + +import org.apache.kafka.common.TopicPartition; + +import java.util.Properties; + + +public class CommittedTimeoutKafkaConsumer { + public static void main(String[] args) { + KafkaConsumer consumer = makeConsumer(); + System.out.println("COMMITTED: " + consumer.committed(new TopicPartition("t", 0))); + } + + public static KafkaConsumer makeConsumer() { + Properties props = new Properties(); + props.put("bootstrap.servers", "okaraman-ld1.linkedin.biz:9092"); + props.put("group.id", System.currentTimeMillis() + ""); + props.put("partition.assignment.strategy", "roundrobin"); + props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + props.put("session.timeout.ms", 30 * 1000); + return new KafkaConsumer<>(props); + } +} \ No newline at end of file diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index 07716876e0766..2f95b465b0ee9 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -1034,7 +1034,7 @@ public OffsetAndMetadata committed(TopicPartition partition) { committed = this.subscriptions.committed(partition); } } else { - Map offsets = coordinator.fetchCommittedOffsets(Collections.singleton(partition)); + Map offsets = coordinator.fetchCommittedOffsets(Collections.singleton(partition), requestTimeoutMs); committed = offsets.get(partition); } @@ -1067,7 +1067,7 @@ public List partitionsFor(String topic) { List parts = cluster.partitionsForTopic(topic); if (parts == null) { metadata.add(topic); - client.awaitMetadataUpdate(); + client.awaitMetadataUpdate(requestTimeoutMs); parts = metadata.fetch().partitionsForTopic(topic); } return parts; diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ListTopicsTimeoutKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ListTopicsTimeoutKafkaConsumer.java new file mode 100644 index 0000000000000..3c3600a327397 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ListTopicsTimeoutKafkaConsumer.java @@ -0,0 +1,36 @@ +/** + * 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.clients.consumer; + +import java.util.Properties; + + +public class ListTopicsTimeoutKafkaConsumer { + public static void main(String[] args) { + KafkaConsumer consumer = makeConsumer(); + System.out.println("LIST TOPICS: " + consumer.listTopics()); + } + + public static KafkaConsumer makeConsumer() { + Properties props = new Properties(); + props.put("bootstrap.servers", "okaraman-ld1.linkedin.biz:9092"); + props.put("group.id", System.currentTimeMillis() + ""); + props.put("partition.assignment.strategy", "roundrobin"); + props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + props.put("session.timeout.ms", 30 * 1000); + props.put("enable.auto.commit", false); + props.put("request.timeout.ms", 20 * 1000); + return new KafkaConsumer<>(props); + } +} \ No newline at end of file diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/PartitionsForTimeoutKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/PartitionsForTimeoutKafkaConsumer.java new file mode 100644 index 0000000000000..31a79de81ae61 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/PartitionsForTimeoutKafkaConsumer.java @@ -0,0 +1,36 @@ +/** + * 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.clients.consumer; + +import java.util.Properties; + + +public class PartitionsForTimeoutKafkaConsumer { + public static void main(String[] args) { + KafkaConsumer consumer = makeConsumer(); + System.out.println("PARTITIONS FOR: " + consumer.partitionsFor("t")); + } + + public static KafkaConsumer makeConsumer() { + Properties props = new Properties(); + props.put("bootstrap.servers", "okaraman-ld1.linkedin.biz:9092"); + props.put("group.id", System.currentTimeMillis() + ""); + props.put("partition.assignment.strategy", "roundrobin"); + props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); + props.put("session.timeout.ms", 30 * 1000); + props.put("enable.auto.commit", false); + props.put("request.timeout.ms", 20 * 1000); + return new KafkaConsumer<>(props); + } +} \ No newline at end of file diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java index 0b611fb1ea216..a518d0dfdbd37 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java @@ -19,6 +19,7 @@ import org.apache.kafka.clients.RequestCompletionHandler; import org.apache.kafka.clients.consumer.ConsumerWakeupException; import org.apache.kafka.common.Node; +import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.requests.AbstractRequest; import org.apache.kafka.common.requests.RequestHeader; @@ -126,6 +127,24 @@ public void awaitMetadataUpdate() { } while (this.metadata.version() == version); } + /** + * Block until the metadata has been refreshed or the timeout has expired. + * @param timeout The maximum duration (in ms) to wait for the request + * @throws TimeoutException if the timeout has expired + */ + public void awaitMetadataUpdate(long timeout) { + long now = time.milliseconds(); + long deadline = now + timeout; + int version = this.metadata.requestUpdate(); + while (this.metadata.version() == version && now < deadline) { + poll(deadline - now, now); + now = time.milliseconds(); + } + + if (this.metadata.version() == version) + throw new TimeoutException("Failed to get metadata after " + timeout + " ms."); + } + /** * Wakeup an active poll. This will cause the polling thread to throw an exception either * on the current poll if one is active, or the next poll. diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Coordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Coordinator.java index 8326549f56e32..00ddb8193e938 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Coordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Coordinator.java @@ -13,6 +13,7 @@ package org.apache.kafka.clients.consumer.internals; import org.apache.kafka.clients.ClientResponse; +import org.apache.kafka.clients.ClientUtils; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.consumer.OffsetCommitCallback; @@ -21,6 +22,7 @@ import org.apache.kafka.common.Node; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.DisconnectException; +import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.UnknownConsumerIdException; import org.apache.kafka.common.metrics.Measurable; import org.apache.kafka.common.metrics.MetricConfig; @@ -125,7 +127,7 @@ public Coordinator(ConsumerNetworkClient client, */ public void refreshCommittedOffsetsIfNeeded() { if (subscriptions.refreshCommitsNeeded()) { - Map offsets = fetchCommittedOffsets(subscriptions.assignedPartitions()); + Map offsets = fetchCommittedOffsets(subscriptions.assignedPartitions(), requestTimeoutMs); for (Map.Entry entry : offsets.entrySet()) { TopicPartition tp = entry.getKey(); // verify assignment is still active @@ -141,22 +143,27 @@ public void refreshCommittedOffsetsIfNeeded() { * @param partitions The partitions to fetch offsets for * @return A map from partition to the committed offset */ - public Map fetchCommittedOffsets(Set partitions) { - while (true) { - ensureCoordinatorKnown(); + public Map fetchCommittedOffsets(Set partitions, long timeout) { + long start = time.milliseconds(); + long deadline = start + timeout; + while (time.milliseconds() < deadline) { + long remainingTime = ClientUtils.maybeGetRemainingTime(start, time.milliseconds(), timeout, ""); + ensureCoordinatorKnown(remainingTime); // contact coordinator to fetch committed offsets RequestFuture> future = sendOffsetFetchRequest(partitions); - client.poll(future); + remainingTime = ClientUtils.maybeGetRemainingTime(start, time.milliseconds(), timeout, ""); + client.poll(future, remainingTime); if (future.succeeded()) return future.value(); - if (!future.isRetriable()) + else if (future.failed() && !future.isRetriable()) throw future.exception(); Utils.sleep(retryBackoffMs); } + throw new TimeoutException("foo"); } /** @@ -231,6 +238,15 @@ public void ensureCoordinatorKnown() { } } + public void ensureCoordinatorKnown(long timeout) { + long now = time.milliseconds(); + long deadline = now + timeout; + while (coordinatorUnknown() && now < deadline) { + RequestFuture future = sendConsumerMetadataRequest(); + client.poll(future, requestTimeoutMs); + } + throw new TimeoutException("foo"); + } @Override public void close() { diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 4608959f01434..8e6b78d5b8b15 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -26,6 +26,7 @@ import org.apache.kafka.common.errors.DisconnectException; import org.apache.kafka.common.errors.InvalidMetadataException; import org.apache.kafka.common.errors.OffsetOutOfRangeException; +import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.stats.Avg; @@ -173,15 +174,15 @@ public void updateFetchPositions(Set partitions) { * @return The map of topics and its partitions */ public Map> getAllTopics(long timeout) { - final HashMap> topicsPartitionInfos = new HashMap<>(); long startTime = time.milliseconds(); while (time.milliseconds() - startTime < timeout) { RequestFuture requestFuture = sendMetadataRequest(); if (requestFuture != null) { - client.poll(requestFuture); + client.poll(requestFuture, timeout); if (requestFuture.succeeded()) { + HashMap> topicsPartitionInfos = new HashMap<>(); MetadataResponse response = new MetadataResponse(requestFuture.value().responseBody()); @@ -190,16 +191,15 @@ public Map> getAllTopics(long timeout) { topic, response.cluster().availablePartitionsForTopic(topic)); return topicsPartitionInfos; - } - - if (!requestFuture.isRetriable()) + } else if (requestFuture.failed() && !requestFuture.isRetriable()) { throw requestFuture.exception(); + } } Utils.sleep(retryBackoffMs); } - return topicsPartitionInfos; + throw new TimeoutException("Failed to get metadata for all topics after " + timeout + " ms."); } /** diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java index b3169d87bbde0..865f308706d44 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java @@ -29,6 +29,7 @@ import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.OffsetOutOfRangeException; +import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.metrics.KafkaMetric; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.protocol.Errors; @@ -330,6 +331,11 @@ public void testGetAllTopics() throws InterruptedException { assertEquals(cluster.topics().size(), allTopics.size()); } + @Test(expected = TimeoutException.class) + public void testGetAllTopicsTimeout() { + fetcher.getAllTopics(0); + } + /* * Send multiple requests. Verify that the client side quota metrics have the right values */