Skip to content

KAFKA-9212; Ensure LeaderAndIsr state updated in controller context during reassignment #7795

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
ijuma merged 3 commits into from
Dec 8, 2019
Merged

KAFKA-9212; Ensure LeaderAndIsr state updated in controller context during reassignment #7795

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f21698b
KAFKA-9212; Update LeaderAndIsr state in controller context after rea…
hachikuji Dec 7, 2019
1616db1
Add Metadata test cases and fix bug handling stale updates
hachikuji Dec 7, 2019
cfb88f4
Fix test and minor tweaks
ijuma Dec 8, 2019
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
26 changes: 11 additions & 15 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 @@ -298,15 +298,10 @@ private MetadataCache handleMetadataResponse(MetadataResponse metadataResponse,
if (metadata.isInternal())
internalTopics.add(metadata.topic());
for (MetadataResponse.PartitionMetadata partitionMetadata : metadata.partitionMetadata()) {

Consumer<PartitionInfo> addToPartitions = partitionInfo -> {
int epoch = partitionMetadata.leaderEpoch().orElse(RecordBatch.NO_PARTITION_LEADER_EPOCH);

@hachikuji hachikuji Dec 7, 2019 •
edited
Loading

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.

Note this was a bug. We were using the leader epoch from the response even if it was stale and we had taken the PartitionInfo from the previous update.

@ijuma ijuma Dec 8, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Do we have a test that covers this bug too?

@stanislavkozlovski stanislavkozlovski Dec 8, 2019

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 believe testStaleMetadata() does that

partitions.add(new MetadataCache.PartitionInfoAndEpoch(partitionInfo, epoch));
};

// Even if the partition's metadata includes an error, we need to handle the update to catch new epochs
// Even if the partition's metadata includes an error, we need to handle
// the update to catch new epochs
updatePartitionInfo(metadata.topic(), partitionMetadata,
metadataResponse.hasReliableLeaderEpochs(), addToPartitions);
metadataResponse.hasReliableLeaderEpochs(), partitions::add);

if (partitionMetadata.error().exception() instanceof InvalidMetadataException) {
log.debug("Requesting metadata update for partition {} due to error {}",
Expand All @@ -332,24 +327,25 @@ private MetadataCache handleMetadataResponse(MetadataResponse metadataResponse,
private void updatePartitionInfo(String topic,
MetadataResponse.PartitionMetadata partitionMetadata,
boolean hasReliableLeaderEpoch,
Consumer<PartitionInfo> partitionInfoConsumer) {
Consumer<MetadataCache.PartitionInfoAndEpoch> partitionInfoConsumer) {
TopicPartition tp = new TopicPartition(topic, partitionMetadata.partition());

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, false)) {
partitionInfoConsumer.accept(MetadataResponse.partitionMetaToInfo(topic, partitionMetadata));
PartitionInfo info = MetadataResponse.partitionMetaToInfo(topic, partitionMetadata);
partitionInfoConsumer.accept(new MetadataCache.PartitionInfoAndEpoch(info, newEpoch));
} else {
// Otherwise ignore the new metadata and use the previously cached info
PartitionInfo previousInfo = cache.cluster().partition(tp);
if (previousInfo != null) {
partitionInfoConsumer.accept(previousInfo);
}
cache.getPartitionInfo(tp).ifPresent(partitionInfoConsumer);
}
} else {
// Handle old cluster formats as well as error responses where leader and epoch are missing
lastSeenLeaderEpochs.remove(tp);
partitionInfoConsumer.accept(MetadataResponse.partitionMetaToInfo(topic, partitionMetadata));
PartitionInfo info = MetadataResponse.partitionMetaToInfo(topic, partitionMetadata);
partitionInfoConsumer.accept(new MetadataCache.PartitionInfoAndEpoch(info,
RecordBatch.NO_PARTITION_LEADER_EPOCH));
}
}

Expand Down
112 changes: 112 additions & 0 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 @@ -23,7 +23,14 @@
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.internals.ClusterResourceListeners;
import org.apache.kafka.common.internals.Topic;
import org.apache.kafka.common.message.MetadataResponseData;
import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseBrokerCollection;
import org.apache.kafka.common.message.MetadataResponseData.MetadataResponsePartition;
import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopic;
import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopicCollection;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.MockTime;
Expand All @@ -33,9 +40,13 @@
import org.junit.Test;

import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

import static org.apache.kafka.test.TestUtils.assertOptional;
import static org.junit.Assert.assertEquals;
Expand Down Expand Up @@ -147,6 +158,107 @@ public void testTimeToNextUpdate_RetryBackoff() {
assertEquals(0, metadata.timeToNextUpdate(now + 1));
}

@Test
public void testIgnoreLeaderEpochInOlderMetadataResponse() {

@stanislavkozlovski stanislavkozlovski Dec 8, 2019

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.

nit: Could we also briefly explain the issue in the tests? Personally, I tend to read tests to understand the expected behavior and the issue with versions earlier than 9 is not immediately apparent

@ijuma ijuma Dec 8, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Is it OK now?

TopicPartition tp = new TopicPartition("topic", 0);

MetadataResponsePartition partitionMetadata = new MetadataResponsePartition()
.setPartitionIndex(tp.partition())
.setLeaderId(5)
.setLeaderEpoch(10)
.setReplicaNodes(Arrays.asList(1, 2, 3))
.setIsrNodes(Arrays.asList(1, 2, 3))
.setOfflineReplicas(Collections.emptyList())
.setErrorCode(Errors.NONE.code());

MetadataResponseTopic topicMetadata = new MetadataResponseTopic()
.setName(tp.topic())
.setErrorCode(Errors.NONE.code())
.setPartitions(Collections.singletonList(partitionMetadata))
.setIsInternal(false);

MetadataResponseTopicCollection topics = new MetadataResponseTopicCollection();
topics.add(topicMetadata);

MetadataResponseData data = new MetadataResponseData()
.setClusterId("clusterId")
.setControllerId(0)
.setTopics(topics)
.setBrokers(new MetadataResponseBrokerCollection());

for (short version = ApiKeys.METADATA.oldestVersion(); version < 9; version++) {
Struct struct = data.toStruct(version);
MetadataResponse response = new MetadataResponse(struct, version);
assertFalse(response.hasReliableLeaderEpochs());
metadata.update(response, 100);
assertTrue(metadata.partitionInfoIfCurrent(tp).isPresent());
MetadataCache.PartitionInfoAndEpoch info = metadata.partitionInfoIfCurrent(tp).get();
assertEquals(-1, info.epoch());
}

for (short version = 9; version <= ApiKeys.METADATA.oldestVersion(); version++) {

@ijuma ijuma Dec 8, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

ApiKeys.METADATA.oldestVersion() -> ApiKeys.METADATA.latestVersion()?

Struct struct = data.toStruct(version);
MetadataResponse response = new MetadataResponse(struct, version);
assertTrue(response.hasReliableLeaderEpochs());
metadata.update(response, 100);
assertTrue(metadata.partitionInfoIfCurrent(tp).isPresent());
MetadataCache.PartitionInfoAndEpoch info = metadata.partitionInfoIfCurrent(tp).get();
assertEquals(10, info.epoch());
}
}

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

MetadataResponsePartition partitionMetadata = new MetadataResponsePartition()
.setPartitionIndex(tp.partition())
.setLeaderId(1)
.setLeaderEpoch(10)
.setReplicaNodes(Arrays.asList(1, 2, 3))
.setIsrNodes(Arrays.asList(1, 2, 3))
.setOfflineReplicas(Collections.emptyList())
.setErrorCode(Errors.NONE.code());

MetadataResponseTopic topicMetadata = new MetadataResponseTopic()
.setName(tp.topic())
.setErrorCode(Errors.NONE.code())
.setPartitions(Collections.singletonList(partitionMetadata))
.setIsInternal(false);

MetadataResponseTopicCollection topics = new MetadataResponseTopicCollection();
topics.add(topicMetadata);

MetadataResponseData data = new MetadataResponseData()
.setClusterId("clusterId")
.setControllerId(0)
.setTopics(topics)
.setBrokers(new MetadataResponseBrokerCollection());

metadata.update(new MetadataResponse(data), 100);

// Older epoch with changed ISR should be ignored
partitionMetadata
.setPartitionIndex(tp.partition())
.setLeaderId(1)
.setLeaderEpoch(9)
.setReplicaNodes(Arrays.asList(1, 2, 3))
.setIsrNodes(Arrays.asList(1, 2))
.setOfflineReplicas(Collections.emptyList())
.setErrorCode(Errors.NONE.code());

metadata.update(new MetadataResponse(data), 101);
assertEquals(Optional.of(10), metadata.lastSeenLeaderEpoch(tp));

assertTrue(metadata.partitionInfoIfCurrent(tp).isPresent());
MetadataCache.PartitionInfoAndEpoch info = metadata.partitionInfoIfCurrent(tp).get();

List<Integer> cachedIsr = Arrays.stream(info.partitionInfo().inSyncReplicas())
.map(Node::id).collect(Collectors.toList());
assertEquals(Arrays.asList(1, 2, 3), cachedIsr);
assertEquals(10, info.epoch());
}

@Test
public void testFailedUpdate() {
long time = 100;
Expand Down