Skip to content
Merged
Changes from 1 commit
Commits
Show all changes
46 commits
Select commit Hold shift + click to select a range
37b5c63
first version
guozhangwang Sep 6, 2019
4a34116
add unit tests
guozhangwang Sep 7, 2019
d964cea
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Sep 10, 2019
f5b8f96
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Sep 11, 2019
b07bac2
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Sep 12, 2019
29f8942
fix unit tests
guozhangwang Sep 12, 2019
3000d6f
add RebalanceInProgressException
guozhangwang Sep 13, 2019
8464fa7
rebase from trunk
guozhangwang Sep 17, 2019
f804b3d
refactor unit tests
guozhangwang Sep 17, 2019
3c62029
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Sep 17, 2019
284ef77
poll with 1ms than 0ms for discover coordinator
guozhangwang Sep 18, 2019
9825063
refactor the newly added test with auto-tick 1ms
guozhangwang Sep 18, 2019
929048a
rebase from trunk
guozhangwang Oct 21, 2019
58adc09
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Oct 22, 2019
c79a93d
fix unit tests
guozhangwang Oct 22, 2019
75c4e88
rebase from trunk again
guozhangwang Oct 22, 2019
95ef3fb
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Oct 27, 2019
3b844ff
use RetriableCommitFailedException
guozhangwang Oct 27, 2019
781514d
unit tests
guozhangwang Oct 27, 2019
8b08a3e
handle retrialbe commit failed in streams
guozhangwang Oct 27, 2019
8fa5d74
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Oct 29, 2019
d75fcb9
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Oct 29, 2019
b176e4e
fix unit tests
guozhangwang Oct 29, 2019
7a826f9
remove imports
guozhangwang Oct 29, 2019
31e372e
fix a real bug exposed by flaky test
guozhangwang Oct 30, 2019
366739b
remove debugging info
guozhangwang Oct 30, 2019
faf787e
minor fix on CommitFailed class
guozhangwang Oct 30, 2019
c71af82
rebase from trunk
guozhangwang Nov 8, 2019
ffe6146
more java doc
guozhangwang Nov 8, 2019
518049c
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Dec 11, 2019
30d5e53
address comments
guozhangwang Dec 12, 2019
fa3456d
address comments
guozhangwang Dec 24, 2019
0220d96
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Jan 2, 2020
27d013a
address comment
guozhangwang Jan 2, 2020
94611cc
do not reset if generation has changed
guozhangwang Jan 2, 2020
b529847
remove restrictive check on commit
guozhangwang Jan 3, 2020
33fd785
check assignment
guozhangwang Jan 3, 2020
1df9808
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Jan 3, 2020
3057405
more comments
guozhangwang Jan 3, 2020
21f531d
remember the fatal exception and throw
guozhangwang Jan 4, 2020
effb29d
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Jan 6, 2020
6fbd976
comments
guozhangwang Jan 6, 2020
f4e7111
Merge branch 'trunk' of https://github.com/apache/kafka into K8421-re…
guozhangwang Jan 8, 2020
91b6606
move back to RebalanceInProgressException
guozhangwang Jan 8, 2020
0ad52ec
address comments
guozhangwang Jan 9, 2020
27f76b7
rebase from trunk
guozhangwang Jan 9, 2020
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
Original file line number Diff line number Diff line change
Expand Up @@ -761,7 +761,11 @@ public boolean refreshCommittedOffsetsIfNeeded(Timer timer) {

log.info("Setting offset for partition {} to the committed offset {}", tp, position);
entry.getValue().leaderEpoch().ifPresent(epoch -> this.metadata.updateLastSeenEpochIfNewer(entry.getKey(), epoch));
this.subscriptions.seekUnvalidated(tp, position);

// it's possible that the partition is no longer assigned when the response is received,

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.

Do you think it's worth adding a debug message saying that we're ignoring the fetched offset?

// so we need to ignore seeking if that's the case
if (this.subscriptions.isAssigned(tp))

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 This is the change I make for auto offset reset.

I thought about adding a unit test but since the logic is wrapped in a single refreshCommittedOffsetsIfNeeded it is hard to change subscription in between without breaking it into multiple ones, but I feel it is too messy to test a singleton function so I did not add one.

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.

Does it make sense to move this check above a little bit?

if (offsetAndMetadata != null && subscriptions.isAssigned(tp)) {

That will make the log message less confusing.

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.

Hmm, but I thought even if the partition is no longer assigned, we may still want to update its epoch; on the other hand your concern is valid. Will tweak it a bit more.

this.subscriptions.seekUnvalidated(tp, position);
}
}
return true;
Expand Down