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 all 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
55 changes: 28 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,35 @@ 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 have seen a newer leader epoch. This is called by the client
* any time it handles a response from the broker that includes leader epoch, except for UpdateMetadata which
* follows a different code path ({@link #update}).
*
* @param topicPartition
* @param leaderEpoch
* @return true if we updated the last seen epoch, false otherwise
*/
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 {} for partition {}", oldEpoch, leaderEpoch, topicPartition);

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) {
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 +192,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 +370,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
9 changes: 5 additions & 4 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 @@ -486,7 +486,7 @@ public void validateOffsetsIfNeeded() {
// Validate each partition against the current leader and epoch
subscriptions.assignedPartitions().forEach(topicPartition -> {
ConsumerMetadata.LeaderAndEpoch leaderAndEpoch = metadata.currentLeader(topicPartition);
subscriptions.maybeValidatePositionForCurrentLeader(topicPartition, leaderAndEpoch);
subscriptions.maybeValidatePositionForCurrentLeader(apiVersions, topicPartition, leaderAndEpoch);
});

// Collect positions needing validation, with backoff
Expand Down Expand Up @@ -756,7 +756,7 @@ public void onFailure(RuntimeException e) {
}
}

private boolean hasUsableOffsetForLeaderEpochVersion(NodeApiVersions nodeApiVersions) {
static boolean hasUsableOffsetForLeaderEpochVersion(NodeApiVersions nodeApiVersions) {
ApiVersion apiVersion = nodeApiVersions.apiVersion(ApiKeys.OFFSET_FOR_LEADER_EPOCH);
if (apiVersion == null)
return false;
Expand Down Expand Up @@ -1101,8 +1101,9 @@ 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 ->
subscriptions.maybeValidatePositionForCurrentLeader(apiVersions, tp, metadata.currentLeader(tp))
);

long currentTimeMs = time.milliseconds();

Expand Down
29 changes: 27 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,9 @@
*/
package org.apache.kafka.clients.consumer.internals;

import org.apache.kafka.clients.ApiVersions;
import org.apache.kafka.clients.Metadata;
import org.apache.kafka.clients.NodeApiVersions;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.NoOffsetForPartitionException;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
Expand Down Expand Up @@ -45,6 +47,8 @@
import java.util.stream.Collector;
import java.util.stream.Collectors;

import static org.apache.kafka.clients.consumer.internals.Fetcher.hasUsableOffsetForLeaderEpochVersion;

/**
* A class for tracking the topics, partitions, and offsets for the consumer. A partition
* is "assigned" either directly with {@link #assignFromUser(Set)} (manual assignment)
Expand Down Expand Up @@ -422,8 +426,29 @@ public synchronized void position(TopicPartition tp, FetchPosition position) {
assignedState(tp).position(position);
}

public synchronized boolean maybeValidatePositionForCurrentLeader(TopicPartition tp, Metadata.LeaderAndEpoch leaderAndEpoch) {
return assignedState(tp).maybeValidatePosition(leaderAndEpoch);
/**
* Enter the offset validation state if the leader for this partition is known to support a usable version of the
* OffsetsForLeaderEpoch API. If the leader node does not support the API, simply complete the offset validation.
*
* @param apiVersions
* @param tp
* @param leaderAndEpoch
* @return true if we enter the offset validation state
*/
public synchronized boolean maybeValidatePositionForCurrentLeader(ApiVersions apiVersions, TopicPartition tp,
Metadata.LeaderAndEpoch leaderAndEpoch) {
if (leaderAndEpoch.leader.isPresent()) {
NodeApiVersions nodeApiVersions = apiVersions.get(leaderAndEpoch.leader.get().idString());
if (nodeApiVersions == null || hasUsableOffsetForLeaderEpochVersion(nodeApiVersions)) {
return assignedState(tp).maybeValidatePosition(leaderAndEpoch);
} else {
// If the broker does not support a newer version of OffsetsForLeaderEpoch, we skip validation
completeValidation(tp);
return false;
}
} else {
return assignedState(tp).maybeValidatePosition(leaderAndEpoch);

@mumrah mumrah May 22, 2020

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 wonder, do we really need this call here? If the leader is not present the epoch shouldn't be present either -- right? If that's the case, then the call to maybeValidatePosition will short circuit

        private boolean maybeValidatePosition(Metadata.LeaderAndEpoch currentLeaderAndEpoch) {
            if (this.fetchState.equals(FetchStates.AWAIT_RESET)) {
                return false;
            }

            if (!currentLeaderAndEpoch.leader.isPresent() && !currentLeaderAndEpoch.epoch.isPresent()) {
                return false;
            }

            if (position != null && !position.currentLeader.equals(currentLeaderAndEpoch)) {
                FetchPosition newPosition = new FetchPosition(position.offset, position.offsetEpoch, currentLeaderAndEpoch);
                validatePosition(newPosition);
                preferredReadReplica = null;
            }
            return this.fetchState.equals(FetchStates.AWAIT_VALIDATION);
        }

@mumrah mumrah May 22, 2020

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.

Oh, actually looking at the javadoc for LeaderAndEpoch, I see

It is also possible that we know of the leader epoch, but not the leader when it is derived from an external source (e.g. a committed offset).

Also in Metadata, we do return a LeaderAndEpoch with the last-seen epoch, but no leader if the metadata is stale. So, I guess it makes sense to keep this call in maybeValidatePositionForCurrentLeader

}
}

/**
Expand Down
73 changes: 51 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 All @@ -325,6 +329,46 @@ public void testRequestUpdate() {
}
}

@Test
public void testUpdateLastEpoch() {
TopicPartition tp = new TopicPartition("topic-1", 0);

MetadataResponse metadataResponse = emptyMetadataResponse();
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 0L);

// if we have no leader epoch, this call shouldn't do anything
assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 0));
assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 1));
assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 2));
assertFalse(metadata.lastSeenLeaderEpoch(tp).isPresent());

// Metadata with newer epoch is handled
metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 10);
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 1L);
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 10));

// Don't update to an older one
assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 1));
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 10));

// Don't cause update if it's the same one
assertFalse(metadata.updateLastSeenEpochIfNewer(tp, 10));
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 10));

// Update if we see newer epoch
assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 12));
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 12));

metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 12);
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 2L);
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 12));

// Don't overwrite metadata with older epoch
metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), Collections.singletonMap("topic-1", 1), _tp -> 11);
metadata.updateWithCurrentRequestVersion(metadataResponse, false, 3L);
assertOptional(metadata.lastSeenLeaderEpoch(tp), leaderAndEpoch -> assertEquals(leaderAndEpoch.intValue(), 12));
}

@Test
public void testRejectOldMetadata() {
Map<String, Integer> partitionCounts = new HashMap<>();
Expand Down Expand Up @@ -378,26 +422,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 +430,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 +478,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
Loading