Skip to content

KAFKA-2632: move fetchable check ahead in handleFetchResponse #295

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
guozhangwang wants to merge 3 commits into from
+18 −6
Closed

KAFKA-2632: move fetchable check ahead in handleFetchResponse #295

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6ef55ee
KAFKA-2632.v1
guozhangwang Oct 11, 2015
48ddf18
address Jason's comments
guozhangwang Oct 12, 2015
9fa1034
some further refactoring
guozhangwang Oct 13, 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
24 changes: 18 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 @@ -302,13 +302,23 @@ public Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords() {
throwIfOffsetOutOfRange();

for (PartitionRecords<K, V> part : this.records) {
if (!subscriptions.isFetchable(part.partition)) {
log.debug("Ignoring fetched records for {} since it is no longer fetchable", part.partition);
if (!subscriptions.isAssigned(part.partition)) {

@hachikuji hachikuji Oct 12, 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.

I think we still need this check. There may be a rebalance between the time a fetch response is handled and the time it is returned to the user.

// this can happen when a rebalance happened before fetched records are returned to the consumer's poll call
log.debug("Not returning fetched records for partition {} since it is no longer assigned", part.partition);
continue;
}

Long consumed = subscriptions.consumed(part.partition);
if (consumed != null && part.fetchOffset == consumed) {
// note that the consumed position should always be available
// as long as the partition is still assigned
long consumed = subscriptions.consumed(part.partition);
if (!subscriptions.isFetchable(part.partition)) {
// this can happen when a partition consumption paused before fetched records are returned to the consumer's poll call
log.debug("Not returning fetched records for assigned partition {} since it is no longer fetchable", part.partition);

// we also need to reset the fetch positions to pretend we did not fetch
// this partition in the previous request at all
subscriptions.fetched(part.partition, consumed);
} else if (part.fetchOffset == consumed) {
List<ConsumerRecord<K, V>> records = drained.get(part.partition);
if (records == null) {
records = part.records;
Expand Down Expand Up @@ -445,8 +455,10 @@ private void handleFetchResponse(ClientResponse resp, FetchRequest request) {
for (Map.Entry<TopicPartition, FetchResponse.PartitionData> entry : response.responseData().entrySet()) {
TopicPartition tp = entry.getKey();
FetchResponse.PartitionData partition = entry.getValue();
if (!subscriptions.assignedPartitions().contains(tp)) {
log.debug("Ignoring fetched data for partition {} which is no longer assigned.", tp);
if (!subscriptions.isFetchable(tp)) {
// this can happen when a rebalance happened or a partition consumption paused
// while fetch is still in-flight
log.debug("Ignoring fetched records for partition {} since it is no longer fetchable", tp);
} else if (partition.errorCode == Errors.NONE.code()) {
long fetchOffset = request.fetchData().get(tp).offset;

Expand Down