Skip to content

KAFKA-9724 Newer clients not always sending fetch request to older brokers #8376

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

Merged
mumrah merged 13 commits into from
May 27, 2020
Merged

KAFKA-9724 Newer clients not always sending fetch request to older brokers #8376

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
c7ba8d5
AWAITING_VALIDATION state causing clients to skip sending fetches to …
mumrah Mar 27, 2020
4a461ec
Don't overwrite a partition's last seen epoch if it wasn't set already
mumrah Apr 7, 2020
349d7e4
Actually implement the new logic
mumrah Apr 7, 2020
8433abb
Add some tests
mumrah Apr 7, 2020
567dc2d
Simply logic for completing validation in prepareFetchRequests
mumrah Apr 7, 2020
48605dc
Feedback from PR
mumrah Apr 7, 2020
ce38b74
checkstyle
mumrah Apr 8, 2020
45c20d8
Merge remote-tracking branch 'apache-github/trunk' into KAFKA-9724
mumrah Apr 8, 2020
c8d4a8c
Add unit test for epoch updates in MetadataTest
mumrah Apr 8, 2020
7a1d6e6
Merge remote-tracking branch 'apache-github/trunk' into KAFKA-9724
mumrah May 18, 2020
e05d8ad
Update javadoc for updateLastSeenEpochIfNewer
mumrah May 18, 2020
aa87a4b
Push logic into maybeValidatePositionForCurrentLeader
mumrah May 22, 2020
37bb3a5
Checkstyle
mumrah May 26, 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
51 changes: 24 additions & 27 deletions clients/src/main/java/org/apache/kafka/clients/Metadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.function.Predicate;
import java.util.function.Supplier;

import static org.apache.kafka.common.record.RecordBatch.NO_PARTITION_LEADER_EPOCH;
Expand Down Expand Up @@ -156,14 +155,31 @@ public synchronized int requestUpdateForNewTopics() {
}

/**
* Request an update for the partition metadata iff the given leader epoch is newer than the last seen leader epoch
* Request an update for the partition metadata iff we encounter a leader epoch that is newer than the last seen leader epoch
Comment thread
mumrah marked this conversation as resolved.
Outdated
*/
public synchronized boolean updateLastSeenEpochIfNewer(TopicPartition topicPartition, int leaderEpoch) {
Objects.requireNonNull(topicPartition, "TopicPartition cannot be null");
if (leaderEpoch < 0)
throw new IllegalArgumentException("Invalid leader epoch " + leaderEpoch + " (must be non-negative)");

boolean updated = updateLastSeenEpoch(topicPartition, leaderEpoch, oldEpoch -> leaderEpoch > oldEpoch);
Integer oldEpoch = lastSeenLeaderEpochs.get(topicPartition);
log.trace("Determining if we should replace existing epoch {} with new epoch {}", oldEpoch, leaderEpoch);
Comment thread
mumrah marked this conversation as resolved.
Outdated

final boolean updated;
if (oldEpoch == null) {
log.debug("Not replacing null epoch with new epoch {} for partition {}", leaderEpoch, topicPartition);
updated = false;
} else {
if (leaderEpoch > oldEpoch) {
Comment thread
mumrah marked this conversation as resolved.
Outdated
log.debug("Updating last seen epoch from {} to {} for partition {}", oldEpoch, leaderEpoch, topicPartition);
lastSeenLeaderEpochs.put(topicPartition, leaderEpoch);
updated = true;
} else {
log.debug("Not replacing existing epoch {} with new epoch {} for partition {}", oldEpoch, leaderEpoch, topicPartition);
updated = false;
}
}

this.needFullUpdate = this.needFullUpdate || updated;
return updated;
}
Expand All @@ -172,29 +188,6 @@ public Optional<Integer> lastSeenLeaderEpoch(TopicPartition topicPartition) {
return Optional.ofNullable(lastSeenLeaderEpochs.get(topicPartition));
}

/**
* Conditionally update the leader epoch for a partition
*
* @param topicPartition topic+partition to update the epoch for
* @param epoch the new epoch
* @param epochTest a predicate to determine if the old epoch should be replaced
* @return true if the epoch was updated, false otherwise
*/
private synchronized boolean updateLastSeenEpoch(TopicPartition topicPartition,

@mumrah mumrah Apr 7, 2020 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I removed this since at this point the two callers have different needs for updating the epoch and adding a flag felt pretty kludgy

int epoch,
Predicate<Integer> epochTest) {
Integer oldEpoch = lastSeenLeaderEpochs.get(topicPartition);
log.trace("Determining if we should replace existing epoch {} with new epoch {}", oldEpoch, epoch);
if (oldEpoch == null || epochTest.test(oldEpoch)) {
log.debug("Updating last seen epoch from {} to {} for partition {}", oldEpoch, epoch, topicPartition);
lastSeenLeaderEpochs.put(topicPartition, epoch);
return true;
} else {
log.debug("Not replacing existing epoch {} with new epoch {} for partition {}", oldEpoch, epoch, topicPartition);
return false;
}
}

/**
* Check whether an update has been explicitly requested.
*
Expand Down Expand Up @@ -373,10 +366,14 @@ private Optional<MetadataResponse.PartitionMetadata> updateLatestMetadata(
if (hasReliableLeaderEpoch && partitionMetadata.leaderEpoch.isPresent()) {
int newEpoch = partitionMetadata.leaderEpoch.get();
// If the received leader epoch is at least the same as the previous one, update the metadata
if (updateLastSeenEpoch(tp, newEpoch, oldEpoch -> newEpoch >= oldEpoch)) {
Integer currentEpoch = lastSeenLeaderEpochs.get(tp);
if (currentEpoch == null || newEpoch >= currentEpoch) {
log.debug("Updating last seen epoch for partition {} from {} to epoch {} from new metadata", tp, currentEpoch, newEpoch);
lastSeenLeaderEpochs.put(tp, newEpoch);
return Optional.of(partitionMetadata);
} else {
// Otherwise ignore the new metadata and use the previously cached info
log.debug("Got metadata for an older epoch {} (current is {}) for partition {}, not updating", newEpoch, currentEpoch, tp);
return cache.partitionMetadata(tp);
}
} else {
Expand Down
16 changes: 14 additions & 2 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 @@ -1098,8 +1098,20 @@ private Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests() {
Map<Node, FetchSessionHandler.Builder> fetchable = new LinkedHashMap<>();

// Ensure the position has an up-to-date leader
subscriptions.assignedPartitions().forEach(
tp -> subscriptions.maybeValidatePositionForCurrentLeader(tp, metadata.currentLeader(tp)));
subscriptions.assignedPartitions().forEach(tp -> {
Metadata.LeaderAndEpoch leaderAndEpoch = metadata.currentLeader(tp);
if (leaderAndEpoch.leader.isPresent()) {
NodeApiVersions nodeApiVersions = apiVersions.get(leaderAndEpoch.leader.get().idString());
Comment thread
mumrah marked this conversation as resolved.
Outdated
if (nodeApiVersions == null || hasUsableOffsetForLeaderEpochVersion(nodeApiVersions)) {
subscriptions.maybeValidatePositionForCurrentLeader(tp, metadata.currentLeader(tp));
} else {
// If the broker does not support a newer version of OffsetsForLeaderEpoch, we skip validation
subscriptions.completeValidation(tp);
}
} else {
subscriptions.maybeValidatePositionForCurrentLeader(tp, metadata.currentLeader(tp));
}
});

long currentTimeMs = time.milliseconds();

Expand Down
33 changes: 11 additions & 22 deletions clients/src/test/java/org/apache/kafka/clients/MetadataTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -313,6 +313,10 @@ public void testRequestUpdate() {
boolean[] updateResult = {true, false, false, false, false, true, false, false, false, true};
TopicPartition tp = new TopicPartition("topic", 0);

MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1,
Collections.emptyMap(), Collections.singletonMap("topic", 1), _tp -> 0);
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 10L);

for (int i = 0; i < epochs.length; i++) {
metadata.updateLastSeenEpochIfNewer(tp, epochs[i]);
if (updateResult[i]) {
Expand Down Expand Up @@ -378,26 +382,6 @@ public void testRejectOldMetadata() {
}
}

@Test
public void testMaybeRequestUpdate() {
Comment thread
mumrah marked this conversation as resolved.
TopicPartition tp = new TopicPartition("topic-1", 0);
metadata.updateWithCurrentRequestVersion(emptyMetadataResponse(), false, 0L);
assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 1));
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 1);

metadata.updateWithCurrentRequestVersion(emptyMetadataResponse(), false, 1L);
assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 1));
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 1);

metadata.updateWithCurrentRequestVersion(emptyMetadataResponse(), false, 2L);
assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 0));
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 1);

metadata.updateWithCurrentRequestVersion(emptyMetadataResponse(), false, 3L);
assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 2));
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 2);
}

@Test
public void testOutOfBandEpochUpdate() {
Map<String, Integer> partitionCounts = new HashMap<>();
Expand All @@ -406,15 +390,15 @@ public void testOutOfBandEpochUpdate() {

metadata.updateWithCurrentRequestVersion(emptyMetadataResponse(), false, 0L);

assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 99));
assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 99));

// Update epoch to 100
MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 100);
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 10L);
assertNotNull(metadata.fetch().partition(tp));
assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);

// Simulate a leader epoch from another response, like a fetch response (not yet implemented)
// Simulate a leader epoch from another response, like a fetch response or list offsets
assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 101));

// Cache of partition stays, but current partition info is not available since it's stale
Expand Down Expand Up @@ -454,6 +438,11 @@ public void testNoEpoch() {
assertTrue(metadata.partitionMetadataIfCurrent(tp).isPresent());
assertEquals(0, metadata.partitionMetadataIfCurrent(tp).get().partition());
assertEquals(Optional.of(0), metadata.partitionMetadataIfCurrent(tp).get().leaderId);

// Since epoch was null, this shouldn't update it
metadata.updateLastSeenEpochIfNewer(tp, 10);
assertTrue(metadata.partitionMetadataIfCurrent(tp).isPresent());
assertFalse(metadata.partitionMetadataIfCurrent(tp).get().leaderEpoch.isPresent());
}

@Test
Expand Down
102 changes: 90 additions & 12 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 @@ -1592,6 +1592,53 @@ public void testResetOffsetsMetadataRefresh() {
assertEquals(5, subscriptions.position(tp0).offset);
}

@Test
public void testListOffsetNoUpdateMissingEpoch() {
buildFetcher();

// Set up metadata with no leader epoch
subscriptions.assignFromUser(singleton(tp0));
MetadataResponse metadataWithNoLeaderEpochs = TestUtils.metadataUpdateWith(
"kafka-cluster", 1, Collections.emptyMap(), singletonMap(topicName, 4), tp -> null);
client.updateMetadata(metadataWithNoLeaderEpochs);

// Return a ListOffsets response with leaderEpoch=1, we should ignore it
subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.LATEST);
client.prepareResponse(listOffsetRequestMatcher(ListOffsetRequest.LATEST_TIMESTAMP),
listOffsetResponse(tp0, Errors.NONE, 1L, 5L, 1));
fetcher.resetOffsetsIfNeeded();
consumerClient.pollNoWakeup();

// Reset should be satisfied and no metadata update requested
assertFalse(subscriptions.isOffsetResetNeeded(tp0));
assertFalse(metadata.updateRequested());
assertFalse(metadata.lastSeenLeaderEpoch(tp0).isPresent());
}

@Test
public void testListOffsetUpdateEpoch() {
buildFetcher();

// Set up metadata with leaderEpoch=1
subscriptions.assignFromUser(singleton(tp0));
MetadataResponse metadataWithLeaderEpochs = TestUtils.metadataUpdateWith(
"kafka-cluster", 1, Collections.emptyMap(), singletonMap(topicName, 4), tp -> 1);
client.updateMetadata(metadataWithLeaderEpochs);

// Reset offsets to trigger ListOffsets call
subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.LATEST);

// Now we see a ListOffsets with leaderEpoch=2 epoch, we trigger a metadata update
client.prepareResponse(listOffsetRequestMatcher(ListOffsetRequest.LATEST_TIMESTAMP, 1),
listOffsetResponse(tp0, Errors.NONE, 1L, 5L, 2));
fetcher.resetOffsetsIfNeeded();
consumerClient.pollNoWakeup();

assertFalse(subscriptions.isOffsetResetNeeded(tp0));
assertTrue(metadata.updateRequested());
assertOptional(metadata.lastSeenLeaderEpoch(tp0), epoch -> assertEquals((long) epoch, 2));
}

@Test
public void testUpdateFetchPositionDisconnect() {
buildFetcher();
Expand Down Expand Up @@ -3620,18 +3667,35 @@ public void testOffsetValidationSkippedForOldBroker() {
apiVersions.update(node.idString(), NodeApiVersions.create(
ApiKeys.OFFSET_FOR_LEADER_EPOCH.id, (short) 0, (short) 2));

// Seek with a position and leader+epoch
Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(
metadata.currentLeader(tp0).leader, Optional.of(epochOne));
subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch));
{
// Seek with a position and leader+epoch
Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(
metadata.currentLeader(tp0).leader, Optional.of(epochOne));
subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch));

// Update metadata to epoch=2, enter validation
metadata.updateWithCurrentRequestVersion(TestUtils.metadataUpdateWith("dummy", 1,
Collections.emptyMap(), partitionCounts, tp -> epochTwo), false, 0L);
fetcher.validateOffsetsIfNeeded();
// Update metadata to epoch=2, enter validation
metadata.updateWithCurrentRequestVersion(TestUtils.metadataUpdateWith("dummy", 1,
Collections.emptyMap(), partitionCounts, tp -> epochTwo), false, 0L);
fetcher.validateOffsetsIfNeeded();

// Offset validation is skipped
assertFalse(subscriptions.awaitingValidation(tp0));
// Offset validation is skipped
assertFalse(subscriptions.awaitingValidation(tp0));
}

{
// Seek with a position and leader+epoch
Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(
metadata.currentLeader(tp0).leader, Optional.of(epochOne));
subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch));

// Update metadata to epoch=2, enter validation
metadata.updateWithCurrentRequestVersion(TestUtils.metadataUpdateWith("dummy", 1,
Collections.emptyMap(), partitionCounts, tp -> epochTwo), false, 0L);

// Subscription should not stay in AWAITING_VALIDATION in prepareFetchRequest
assertEquals(1, fetcher.sendFetches());
assertFalse(subscriptions.awaitingValidation(tp0));
}
}

@Test
Expand Down Expand Up @@ -4012,13 +4076,27 @@ private MockClient.RequestMatcher listOffsetRequestMatcher(final long timestamp)
};
}

private MockClient.RequestMatcher listOffsetRequestMatcher(final long timestamp, final int leaderEpoch) {
// matches any list offset request with the provided timestamp
return body -> {
ListOffsetRequest req = (ListOffsetRequest) body;
return req.partitionTimestamps().equals(Collections.singletonMap(
tp0, new ListOffsetRequest.PartitionData(timestamp, Optional.of(leaderEpoch))));
};
}

private ListOffsetResponse listOffsetResponse(Errors error, long timestamp, long offset) {
return listOffsetResponse(tp0, error, timestamp, offset);
}

private ListOffsetResponse listOffsetResponse(TopicPartition tp, Errors error, long timestamp, long offset) {
ListOffsetResponse.PartitionData partitionData = new ListOffsetResponse.PartitionData(error, timestamp, offset,
Optional.empty());
return listOffsetResponse(tp, error, timestamp, offset, null);
}

private ListOffsetResponse listOffsetResponse(TopicPartition tp, Errors error, long timestamp, long offset,
Integer leaderEpoch) {
ListOffsetResponse.PartitionData partitionData = new ListOffsetResponse.PartitionData(
error, timestamp, offset, Optional.ofNullable(leaderEpoch));
Map<TopicPartition, ListOffsetResponse.PartitionData> allPartitionData = new HashMap<>();
allPartitionData.put(tp, partitionData);
return new ListOffsetResponse(allPartitionData);
Expand Down