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 2 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
15 changes: 12 additions & 3 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 @@ -303,7 +303,14 @@ public Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords() {

for (PartitionRecords<K, V> part : this.records) {
if (!subscriptions.isFetchable(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.

log.debug("Ignoring fetched records for {} since it is no longer fetchable", part.partition);
// this can happen when a rebalance happened or a partition consumption paused
// before fetched records are returned to the consumer's poll call
log.debug("Not returning fetched records for 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, part.records.get(0).offset());

@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.

Isn't it a little dangerous to assume that the fetch position in the returned record is still relevant? Suppose we have the following sequence:

  1. User calls poll() and we initiate a fetch.
  2. User calls commitSync() (or some other blocking operation) and the fetch returns.
  3. User calls seek() with a new offset.
  4. User calls poll() and now we've written over their seeked offset.

Maybe instead we should reset the fetched position to the consumed position?

@guozhangwang guozhangwang Oct 13, 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.

@hachikuji I thought about this, but consumed() may return null for the first time so we cannot rely on that value either. Maybe we should just combine these two? I.e.:

If consumed() != null, set to consumed;
otherwise this MUST be the first fetch ever returned before any seek() was called, hence resetting to first offset should be safe.


continue;
}

Expand Down Expand Up @@ -445,8 +452,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