Skip to content

KAFKA-9261; Client should handle unavailable leader metadata #7770

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
hachikuji merged 12 commits into from
Feb 5, 2020
Merged

KAFKA-9261; Client should handle unavailable leader metadata #7770

Show file tree
Hide file tree
Changes from 11 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
68db4c7
Use PartitionMetadata and delete PartitionInfoAndEpoch
hachikuji Dec 3, 2019
c7d6871
Rip out references to Node in PartitionMetadata
hachikuji Dec 3, 2019
2d80cf9
Loosen strict leader node requirement
hachikuji Dec 3, 2019
5b350f7
Unroll flatMap to avoid spotbugs warning
hachikuji Dec 4, 2019
7548821
Address review comments
hachikuji Jan 27, 2020
8dcf8d8
Add test case from 2.4 patch
hachikuji Feb 3, 2020
4c43b75
Review comments
hachikuji Feb 3, 2020
bc9a118
Add hasAliveEndpoint to avoid wasted Option creation
hachikuji Feb 3, 2020
c812647
Make leaderId Optional in PartitionMetadata
hachikuji Feb 3, 2020
f818a2c
Fix checkstyle
hachikuji Feb 4, 2020
31e6632
Fix failing tests
hachikuji Feb 4, 2020
fb5a99f
A few cleanups in Metadata
hachikuji Feb 5, 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
80 changes: 46 additions & 34 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 @@ -19,14 +19,12 @@
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.InvalidMetadataException;
import org.apache.kafka.common.errors.InvalidTopicException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.internals.ClusterResourceListeners;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
import org.apache.kafka.common.utils.LogContext;
Expand All @@ -47,6 +45,8 @@
import java.util.function.Predicate;
import java.util.function.Supplier;

import static org.apache.kafka.common.record.RecordBatch.NO_PARTITION_LEADER_EPOCH;

/**
* A class encapsulating some of the logic around metadata.
* <p>
Expand Down Expand Up @@ -151,6 +151,8 @@ public synchronized int requestUpdate() {
*/
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)");
return updateLastSeenEpoch(topicPartition, leaderEpoch, oldEpoch -> leaderEpoch > oldEpoch, true);
}

Expand Down Expand Up @@ -199,16 +201,29 @@ public synchronized boolean updateRequested() {
/**
* Return the cached partition info if it exists and a newer leader epoch isn't known about.
*/
public synchronized Optional<MetadataCache.PartitionInfoAndEpoch> partitionInfoIfCurrent(TopicPartition topicPartition) {
synchronized Optional<MetadataResponse.PartitionMetadata> partitionMetadataIfCurrent(TopicPartition topicPartition) {
Integer epoch = lastSeenLeaderEpochs.get(topicPartition);
Comment thread
ijuma marked this conversation as resolved.
Outdated
Optional<MetadataResponse.PartitionMetadata> partitionMetadata = cache.partitionMetadata(topicPartition);
if (epoch == null) {
// old cluster format (no epochs)
return cache.getPartitionInfo(topicPartition);
return partitionMetadata;
} else {
return cache.getPartitionInfoHavingEpoch(topicPartition, epoch);
return partitionMetadata.filter(metadata ->
metadata.leaderEpoch.orElse(NO_PARTITION_LEADER_EPOCH).equals(epoch));
}
}

public synchronized LeaderAndEpoch currentLeader(TopicPartition topicPartition) {
Optional<MetadataResponse.PartitionMetadata> maybeMetadata = partitionMetadataIfCurrent(topicPartition);
if (!maybeMetadata.isPresent())
return new LeaderAndEpoch(Optional.empty(), Optional.ofNullable(lastSeenLeaderEpochs.get(topicPartition)));

MetadataResponse.PartitionMetadata partitionMetadata = maybeMetadata.get();
Optional<Integer> leaderEpoch = partitionMetadata.leaderEpoch;
Comment thread
ijuma marked this conversation as resolved.
Outdated
Optional<Node> leaderNodeOpt = partitionMetadata.leaderId.flatMap(cache::nodeById);
return new LeaderAndEpoch(leaderNodeOpt, leaderEpoch);
}

public synchronized void bootstrap(List<InetSocketAddress> addresses) {
this.needUpdate = true;
this.updateVersion += 1;
Expand Down Expand Up @@ -245,7 +260,7 @@ public synchronized void update(int requestVersion, MetadataResponse response, l
this.lastSuccessfulRefreshMs = now;
this.updateVersion += 1;

String previousClusterId = cache.cluster().clusterResource().clusterId();
String previousClusterId = cache.clusterResource().clusterId();

this.cache = handleMetadataResponse(response, topic -> retainTopic(topic.topic(), topic.isInternal(), now));

Expand All @@ -254,11 +269,11 @@ public synchronized void update(int requestVersion, MetadataResponse response, l

this.lastSeenLeaderEpochs.keySet().removeIf(tp -> !retainTopic(tp.topic(), false, now));

String newClusterId = cache.cluster().clusterResource().clusterId();
String newClusterId = cache.clusterResource().clusterId();
if (!Objects.equals(previousClusterId, newClusterId)) {
log.info("Cluster ID: {}", newClusterId);
}
clusterResourceListeners.onUpdate(cache.cluster().clusterResource());
clusterResourceListeners.onUpdate(cache.clusterResource());

log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.cache);
}
Expand Down Expand Up @@ -289,7 +304,7 @@ private void checkUnauthorizedTopics(Cluster cluster) {
private MetadataCache handleMetadataResponse(MetadataResponse metadataResponse,
Predicate<MetadataResponse.TopicMetadata> topicsToRetain) {
Set<String> internalTopics = new HashSet<>();
List<MetadataCache.PartitionInfoAndEpoch> partitions = new ArrayList<>();
List<MetadataResponse.PartitionMetadata> partitions = new ArrayList<>();
Comment thread
ijuma marked this conversation as resolved.
Outdated
for (MetadataResponse.TopicMetadata metadata : metadataResponse.topicMetadata()) {
if (!topicsToRetain.test(metadata))
continue;
Expand All @@ -303,9 +318,9 @@ private MetadataCache handleMetadataResponse(MetadataResponse metadataResponse,
updatePartitionInfo(metadata.topic(), partitionMetadata,
metadataResponse.hasReliableLeaderEpochs(), partitions::add);

if (partitionMetadata.error().exception() instanceof InvalidMetadataException) {
if (partitionMetadata.error.exception() instanceof InvalidMetadataException) {
log.debug("Requesting metadata update for partition {} due to error {}",
new TopicPartition(metadata.topic(), partitionMetadata.partition()), partitionMetadata.error());
partitionMetadata.topicPartition, partitionMetadata.error);
requestUpdate();
}
}
Expand All @@ -315,10 +330,13 @@ private MetadataCache handleMetadataResponse(MetadataResponse metadataResponse,
}
}

return new MetadataCache(metadataResponse.clusterId(), new ArrayList<>(metadataResponse.brokers()), partitions,
return new MetadataCache(metadataResponse.clusterId(),
metadataResponse.brokersById(),
partitions,
metadataResponse.topicsByError(Errors.TOPIC_AUTHORIZATION_FAILED),
metadataResponse.topicsByError(Errors.INVALID_TOPIC_EXCEPTION),
internalTopics, metadataResponse.controller());
internalTopics,
metadataResponse.controller());
}

/**
Expand All @@ -327,25 +345,22 @@ private MetadataCache handleMetadataResponse(MetadataResponse metadataResponse,
private void updatePartitionInfo(String topic,
MetadataResponse.PartitionMetadata partitionMetadata,
boolean hasReliableLeaderEpoch,
Consumer<MetadataCache.PartitionInfoAndEpoch> partitionInfoConsumer) {
Consumer<MetadataResponse.PartitionMetadata> partitionInfoConsumer) {
Comment thread
ijuma marked this conversation as resolved.
Outdated
TopicPartition tp = new TopicPartition(topic, partitionMetadata.partition());

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

Expand Down Expand Up @@ -491,23 +506,20 @@ private MetadataRequestAndVersion(MetadataRequest.Builder requestBuilder,
}
}

public synchronized LeaderAndEpoch leaderAndEpoch(TopicPartition tp) {
return partitionInfoIfCurrent(tp)
.map(infoAndEpoch -> {
Node leader = infoAndEpoch.partitionInfo().leader();
return new LeaderAndEpoch(leader == null ? Node.noNode() : leader, Optional.of(infoAndEpoch.epoch()));
})
.orElse(new LeaderAndEpoch(Node.noNode(), lastSeenLeaderEpoch(tp)));
}

/**
* Represents current leader state known in metadata. It is possible that we know the leader, but not the
* epoch if the metadata is received from a broker which does not support a sufficient Metadata API version.
* 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).
*/
public static class LeaderAndEpoch {

public static final LeaderAndEpoch NO_LEADER_OR_EPOCH = new LeaderAndEpoch(Node.noNode(), Optional.empty());
public static final LeaderAndEpoch NO_LEADER_OR_EPOCH = new LeaderAndEpoch(Optional.empty(), Optional.empty());
Comment thread
ijuma marked this conversation as resolved.
Outdated

public final Node leader;
public final Optional<Node> leader;
public final Optional<Integer> epoch;

public LeaderAndEpoch(Node leader, Optional<Integer> epoch) {
public LeaderAndEpoch(Optional<Node> leader, Optional<Integer> epoch) {
this.leader = Objects.requireNonNull(leader);
this.epoch = Objects.requireNonNull(epoch);
}
Expand Down
105 changes: 35 additions & 70 deletions clients/src/main/java/org/apache/kafka/clients/MetadataCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,18 +17,18 @@
package org.apache.kafka.clients;

import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.ClusterResource;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.requests.MetadataResponse;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
Expand All @@ -39,33 +39,33 @@
*/
public class MetadataCache {
private final String clusterId;
private final List<Node> nodes;
private final Map<Integer, Node> nodes;
private final Set<String> unauthorizedTopics;
private final Set<String> invalidTopics;
private final Set<String> internalTopics;
private final Node controller;
private final Map<TopicPartition, PartitionInfoAndEpoch> metadataByPartition;
private final Map<TopicPartition, MetadataResponse.PartitionMetadata> metadataByPartition;

private Cluster clusterInstance;

MetadataCache(String clusterId,
List<Node> nodes,
Collection<PartitionInfoAndEpoch> partitions,
Map<Integer, Node> nodes,
Collection<MetadataResponse.PartitionMetadata> partitions,
Set<String> unauthorizedTopics,
Set<String> invalidTopics,
Set<String> internalTopics,
Node controller) {
this(clusterId, nodes, partitions, unauthorizedTopics, invalidTopics, internalTopics, controller, null);
}

MetadataCache(String clusterId,
List<Node> nodes,
Collection<PartitionInfoAndEpoch> partitions,
Set<String> unauthorizedTopics,
Set<String> invalidTopics,
Set<String> internalTopics,
Node controller,
Cluster clusterInstance) {
private MetadataCache(String clusterId,
Map<Integer, Node> nodes,
Collection<MetadataResponse.PartitionMetadata> partitions,
Set<String> unauthorizedTopics,
Set<String> invalidTopics,
Set<String> internalTopics,
Node controller,
Cluster clusterInstance) {
this.clusterId = clusterId;
this.nodes = nodes;
this.unauthorizedTopics = unauthorizedTopics;
Expand All @@ -74,8 +74,8 @@ public class MetadataCache {
this.controller = controller;

this.metadataByPartition = new HashMap<>(partitions.size());
for (PartitionInfoAndEpoch p : partitions) {
this.metadataByPartition.put(new TopicPartition(p.partitionInfo().topic(), p.partitionInfo().partition()), p);
for (MetadataResponse.PartitionMetadata p : partitions) {
this.metadataByPartition.put(p.topicPartition, p);
}

if (clusterInstance == null) {
Expand All @@ -85,16 +85,12 @@ public class MetadataCache {
}
}

/**
* Return the cached PartitionInfo iff it was for the given epoch
*/
Optional<PartitionInfoAndEpoch> getPartitionInfoHavingEpoch(TopicPartition topicPartition, int epoch) {
PartitionInfoAndEpoch infoAndEpoch = metadataByPartition.get(topicPartition);
return Optional.ofNullable(infoAndEpoch).filter(infoEpoch -> infoEpoch.epoch() == epoch);
Optional<MetadataResponse.PartitionMetadata> partitionMetadata(TopicPartition topicPartition) {
return Optional.ofNullable(metadataByPartition.get(topicPartition));
}

Optional<PartitionInfoAndEpoch> getPartitionInfo(TopicPartition topicPartition) {
return Optional.ofNullable(metadataByPartition.get(topicPartition));
Optional<Node> nodeById(int id) {
return Optional.ofNullable(nodes.get(id));
}

Cluster cluster() {
Expand All @@ -105,25 +101,33 @@ Cluster cluster() {
}
}

ClusterResource clusterResource() {
return new ClusterResource(clusterId);
}

private void computeClusterView() {
List<PartitionInfo> partitionInfos = metadataByPartition.values()
.stream()
.map(PartitionInfoAndEpoch::partitionInfo)
.map(metadata -> MetadataResponse.toPartitionInfo(metadata, nodes))
.collect(Collectors.toList());
this.clusterInstance = new Cluster(clusterId, nodes, partitionInfos, unauthorizedTopics, invalidTopics, internalTopics, controller);
this.clusterInstance = new Cluster(clusterId, nodes.values(), partitionInfos, unauthorizedTopics,
invalidTopics, internalTopics, controller);
}

static MetadataCache bootstrap(List<InetSocketAddress> addresses) {
List<Node> nodes = new ArrayList<>();
Map<Integer, Node> nodes = new HashMap<>();
int nodeId = -1;
for (InetSocketAddress address : addresses)
nodes.add(new Node(nodeId--, address.getHostString(), address.getPort()));
for (InetSocketAddress address : addresses) {
nodes.put(nodeId, new Node(nodeId, address.getHostString(), address.getPort()));
nodeId--;
}
return new MetadataCache(null, nodes, Collections.emptyList(),
Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Cluster.bootstrap(addresses));
Collections.emptySet(), Collections.emptySet(), Collections.emptySet(),
null, Cluster.bootstrap(addresses));
}

static MetadataCache empty() {
return new MetadataCache(null, Collections.emptyList(), Collections.emptyList(),
return new MetadataCache(null, Collections.emptyMap(), Collections.emptyList(),
Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Cluster.empty());
}

Expand All @@ -137,43 +141,4 @@ public String toString() {
'}';
}

public static class PartitionInfoAndEpoch {
private final PartitionInfo partitionInfo;
private final int epoch;

PartitionInfoAndEpoch(PartitionInfo partitionInfo, int epoch) {
this.partitionInfo = partitionInfo;
this.epoch = epoch;
}

public PartitionInfo partitionInfo() {
return partitionInfo;
}

public int epoch() {
return epoch;
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
PartitionInfoAndEpoch that = (PartitionInfoAndEpoch) o;
return epoch == that.epoch &&
Objects.equals(partitionInfo, that.partitionInfo);
}

@Override
public int hashCode() {
return Objects.hash(partitionInfo, epoch);
}

@Override
public String toString() {
return "PartitionInfoAndEpoch{" +
"partitionInfo=" + partitionInfo +
", epoch=" + epoch +
'}';
}
}
}
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1055,7 +1055,7 @@ public void handleSuccessfulResponse(RequestHeader requestHeader, long now, Meta
// This could be a transient issue if listeners were added dynamically to brokers.
List<TopicPartition> missingListenerPartitions = response.topicMetadata().stream().flatMap(topicMetadata ->
topicMetadata.partitionMetadata().stream()
.filter(partitionMetadata -> partitionMetadata.error() == Errors.LISTENER_NOT_FOUND)
.filter(partitionMetadata -> partitionMetadata.error == Errors.LISTENER_NOT_FOUND)
.map(partitionMetadata -> new TopicPartition(topicMetadata.topic(), partitionMetadata.partition())))
.collect(Collectors.toList());
if (!missingListenerPartitions.isEmpty()) {
Expand Down
4 changes: 2 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/admin/internals/MetadataOperationContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,8 +83,8 @@ public Collection<String> topics() {
public static void handleMetadataErrors(MetadataResponse response) {
for (TopicMetadata tm : response.topicMetadata()) {
for (PartitionMetadata pm : tm.partitionMetadata()) {
if (shouldRefreshMetadata(pm.error())) {
throw pm.error().exception();
if (shouldRefreshMetadata(pm.error)) {
throw pm.error.exception();
}
}
}
Expand Down
Loading