《Java 消息中间件·Kafka 内核与客户端》5.3 客户端的元数据刷新

unzip -p ~/.m2/repository/org/apache/kafka/kafka-clients/4.3.0/kafka-clients-4.3.0-sources.jar \
  org/apache/kafka/clients/Metadata.java > /tmp/Metadata.java

public class Metadata implements Closeable {
    private final Logger log;
    private final ExponentialBackoff refreshBackoff;
    private final long metadataExpireMs;
    private int updateVersion;  // bumped on every metadata response
    private int requestVersion; // bumped on every new topic addition
    private long lastRefreshMs;
    private long lastSuccessfulRefreshMs;
    private long attempts;
    private KafkaException fatalException;
    private Set<String> invalidTopics;
    private Set<String> unauthorizedTopics;
    private volatile MetadataSnapshot metadataSnapshot = MetadataSnapshot.empty();
    private boolean needFullUpdate;
    private boolean needPartialUpdate;
    private long equivalentResponseCount;
    private final ClusterResourceListeners clusterResourceListeners;
    private boolean isClosed;
    private final Map<TopicPartition, Integer> lastSeenLeaderEpochs;

public synchronized long timeToNextUpdate(long nowMs) {
    long timeToExpire = updateRequested() ? 0 : Math.max(this.lastSuccessfulRefreshMs + this.metadataExpireMs - nowMs, 0);
    return Math.max(timeToExpire, timeToAllowUpdate(nowMs));
}

public synchronized long timeToAllowUpdate(long nowMs) {
    // Calculate the backoff for attempts which acts when metadata responses fail
    long backoffForAttempts = Math.max(this.lastRefreshMs +
            this.refreshBackoff.backoff(this.attempts > 0 ? this.attempts - 1 : 0) - nowMs, 0);

    // Periodic updates based on expiration resets the equivalent response count so exponential backoff is not used
    if (Math.max(this.lastSuccessfulRefreshMs + this.metadataExpireMs - nowMs, 0) == 0) {
        this.equivalentResponseCount = 0;
    }

    // Calculate the backoff for equivalent responses which acts when metadata responses are not making progress
    long backoffForEquivalentResponseCount = Math.max(this.lastRefreshMs +
            (this.equivalentResponseCount > 0 ? this.refreshBackoff.backoff(this.equivalentResponseCount - 1) : 0) - nowMs, 0);

    return Math.max(backoffForAttempts, backoffForEquivalentResponseCount);
}

this.refreshBackoff = new ExponentialBackoff(
    refreshBackoffMs,
    CommonClientConfigs.RETRY_BACKOFF_EXP_BASE,
    refreshBackoffMaxMs,
    CommonClientConfigs.RETRY_BACKOFF_JITTER);

public synchronized int requestUpdate(final boolean resetEquivalentResponseBackoff) {
    this.needFullUpdate = true;
    if (resetEquivalentResponseBackoff) {
        this.equivalentResponseCount = 0;
    }
    return this.updateVersion;
}

public synchronized int requestUpdateForNewTopics() {
    // Override the timestamp of last refresh to let immediate update.
    this.lastRefreshMs = 0;
    this.needPartialUpdate = true;
    this.equivalentResponseCount = 0;
    this.requestVersion++;
    return this.updateVersion;
}

public synchronized void update(int requestVersion, MetadataResponse response, boolean isPartialUpdate, long nowMs) {
    Objects.requireNonNull(response, "Metadata response cannot be null");
    if (isClosed())
        throw new IllegalStateException("Update requested after metadata close");

    this.needPartialUpdate = requestVersion < this.requestVersion;
    this.lastRefreshMs = nowMs;
    this.attempts = 0;
    this.updateVersion += 1;
    if (!isPartialUpdate) {
        this.needFullUpdate = false;
        this.lastSuccessfulRefreshMs = nowMs;
    }
    this.equivalentResponseCount++;

    String previousClusterId = metadataSnapshot.clusterResource().clusterId();
    this.metadataSnapshot = handleMetadataResponse(response, isPartialUpdate, nowMs);
    // ... 省略错误处理与 listener 通知
    log.debug("Updated cluster metadata updateVersion {} to {}", this.updateVersion, this.metadataSnapshot);
}

public synchronized void failedUpdate(long now) {
    this.lastRefreshMs = now;
    this.attempts++;
    this.equivalentResponseCount = 0;
}

public long maybeUpdate(long now) {
    // should we update our metadata?
    long timeToNextMetadataUpdate = metadata.timeToNextUpdate(now);
    long waitForMetadataFetch = hasFetchInProgress() ? defaultRequestTimeoutMs : 0;

    long metadataTimeout = Math.max(timeToNextMetadataUpdate, waitForMetadataFetch);
    if (metadataTimeout > 0) {
        return metadataTimeout;
    }
    // ... 选 leastLoadedNode，发 Metadata 请求
}

@Override
public synchronized boolean retainTopic(String topic, boolean isInternal, long nowMs) {
    Long expireMs = topics.get(topic);
    if (expireMs == null) {
        return false;
    } else if (newTopics.contains(topic)) {
        return true;
    } else if (expireMs <= nowMs) {
        log.debug("Removing unused topic {} from the metadata list, expiryMs {} now {}", topic, expireMs, nowMs);
        topics.remove(topic);
        return false;
    } else {
        return true;
    }
}

public synchronized void awaitUpdate(final int lastVersion, final long timeoutMs) throws InterruptedException {
    long currentTimeMs = time.milliseconds();
    long deadlineMs = currentTimeMs + timeoutMs < 0 ? Long.MAX_VALUE : currentTimeMs + timeoutMs;
    time.waitObject(this, () -> {
        // Throw fatal exceptions, if there are any.
        maybeThrowFatalException();
        // ...
    }, deadlineMs);
}

java -Dorg.slf4j.simpleLogger.defaultLogLevel=debug \
  -cp "target/classes:$(cat cp.txt)" lab.MetaRefreshLab a2.refresh 2000 10000 8000

Resetting the last seen epoch of partition a2.acks0-0 to 2 since the associated topicId changed from null to xetYQDXzRNWqPRc96l_p4Q

public synchronized Optional<MetadataResponse.PartitionMetadata> partitionMetadataIfCurrent(TopicPartition topicPartition) {
    Integer epoch = lastSeenLeaderEpochs.get(topicPartition);
    Optional<MetadataResponse.PartitionMetadata> partitionMetadata = metadataSnapshot.partitionMetadata(topicPartition);
    if (epoch == null) {
        // old cluster format (no epochs)
        return partitionMetadata;
    } else {
        return partitionMetadata.filter(metadata ->
                metadata.leaderEpoch.orElse(NO_PARTITION_LEADER_EPOCH).equals(epoch));
    }
}