Skip to content

KAFKA-2391 [WIP]: add timeout to KafkaConsumer blocking calls #283

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
onurkaraman wants to merge 3 commits into from
Closed

KAFKA-2391 [WIP]: add timeout to KafkaConsumer blocking calls #283

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e46eb88
add timeout to KafkaConsumer.listTopics
onurkaraman Oct 5, 2015
eb62d4b
add timeout to KafkaConsumer.partitionsFor
onurkaraman Oct 6, 2015
cfad95d
wip
onurkaraman Oct 9, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1067,7 +1067,7 @@ public List<PartitionInfo> partitionsFor(String topic) {
List<PartitionInfo> parts = cluster.partitionsForTopic(topic);
if (parts == null) {
metadata.add(topic);
client.awaitMetadataUpdate();
client.awaitMetadataUpdate(requestTimeoutMs);
parts = metadata.fetch().partitionsForTopic(topic);
}
return parts;
Expand Down
19 changes: 19 additions & 0 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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) {

@ijuma ijuma Oct 7, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One question: what is our plan for cases where the metadata update fails? A few options that I can think of:

  1. Attempt it again until the timeout expires
  2. Throw an exception immediately
  3. Wait until the timeout expires and then throw an exception

If I understand correctly, we are doing 3 at the moment (please correct me if I am wrong). I think 1 or 2 would be better although perhaps more work. We can talk about implementation issues after we decide what we want to do though. cc @hachikuji

@hachikuji hachikuji Oct 7, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The tough part about the consumer is that the handling is context-dependent. If the call to awaitMetadataUpdate happens in KafkaConsumer.poll(), we might want to ignore the timeout and just return an empty record set to the user. If it's a blocking call, we want to throw the exception. I wonder if it would be a good idea to let this variant return a boolean indicating whether the update happened so that the caller can more easily choose what to do?

@onurkaraman onurkaraman Oct 7, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the quick reply @ijuma. Yeah the point you raise is valid.

It looks like KafkaConsumer.listTopics (more specifically, its call to Fetcher.getAllTopics) in this RB is actually doing 1, but KafkaConsumer.partitionsFor (more specifically, its call to ConsumerNetworkClient.awaitMetadataUpdate) is doing 3. Maybe we should make them both act the same.

Also just a heads up, this PR is very much WIP. It only has listTopics and partitionsFor with timeouts so far. I put this PR up partially so that it shows the flaw in using request.timeout.ms as the timeout. From the jira ticket:

  • if request.timeout.ms < session.timeout.ms, then a user may drop JoinGroupResponses which can arrive worst case after session.timeout.ms.
  • if request.timeout.ms > session.timeout.ms, then a user calling one of these blocking calls will lose their session and trigger a rebalance.

@ijuma ijuma Oct 7, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, good points @hachikuji and @onurkaraman. By the way, I totally understand that this is a WIP. I just thought it would be worth discussing this sooner rather than later. :)

It does seem that a variant that lets the caller decide what to do would be good to have. Not sure if a boolean is good enough to represent the 3 possible states (or if the callers need to know about the 3 possible states).

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.
Expand Down
12 changes: 6 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -173,15 +174,15 @@ public void updateFetchPositions(Set<TopicPartition> partitions) {
* @return The map of topics and its partitions
*/
public Map<String, List<PartitionInfo>> getAllTopics(long timeout) {
final HashMap<String, List<PartitionInfo>> topicsPartitionInfos = new HashMap<>();
long startTime = time.milliseconds();

while (time.milliseconds() - startTime < timeout) {
RequestFuture<ClientResponse> requestFuture = sendMetadataRequest();
if (requestFuture != null) {
client.poll(requestFuture);
client.poll(requestFuture, timeout);

if (requestFuture.succeeded()) {
HashMap<String, List<PartitionInfo>> topicsPartitionInfos = new HashMap<>();
MetadataResponse response =
new MetadataResponse(requestFuture.value().responseBody());

Expand All @@ -190,16 +191,15 @@ public Map<String, List<PartitionInfo>> 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.");
}

/**
Expand Down
6 changes: 6 additions & 0 deletions clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
*/
Expand Down