{
  "type": "header",
  "name": "RequestHeader",
  // Version 0 was removed in Apache Kafka 4.0, Version 1 is the new baseline.
  //
  // Version 0 of the RequestHeader is only used by v0 of ControlledShutdownRequest.
  //
  // Version 1 is the first version with ClientId.
  //
  // Version 2 is the first flexible version.
  "validVersions": "1-2",
  "flexibleVersions": "2+",
  "fields": [
    { "name": "RequestApiKey", "type": "int16", "versions": "0+",
      "about": "The API key of this request." },
    { "name": "RequestApiVersion", "type": "int16", "versions": "0+",
      "about": "The API version of this request." },
    { "name": "CorrelationId", "type": "int32", "versions": "0+",
      "about": "The correlation ID of this request." },

    // The ClientId string must be serialized with the old-style two-byte length prefix.
    // The reason is that older brokers must be able to read the request header for any
    // ApiVersionsRequest, even if it is from a newer version.
    // Since the client is sending the ApiVersionsRequest in order to discover what
    // versions are supported, the client does not know the best version to use.
    { "name": "ClientId", "type": "string", "versions": "1+", "nullableVersions": "1+", "flexibleVersions": "none",
      "about": "The client ID string." },
    { "name": "ConsistencyState", "type": "ConsistencyState", "versions": "2+", "taggedVersions": "2+", "tag": 0,
      "about": "Consistency context for the request.", "fields": [
      { "name": "ClusterId", "type": "string", "versions": "2+", "nullableVersions": "2+", "default": "null",
        "about": "The clusterId if known. This is used to validate request against the expected cluster." },
      { "name": "ConsistencyToken", "type": "int64", "versions": "2+", "default": "-1",
        "about": "The latest consistency token seen by the client." }
    ]}
  ]
}

{
  "type": "header",
  "name": "ResponseHeader",
  // Version 1 is the first flexible version.
  "validVersions": "0-1",
  "flexibleVersions": "1+",
  "fields": [
    { "name": "CorrelationId", "type": "int32", "versions": "0+",
      "about": "The correlation ID of this response." },
    { "name": "ConsistencyState", "type": "ConsistencyState", "versions": "1+", "taggedVersions": "1+", "tag": 0,
      "about": "Consistency context for the request.", "fields": [
      { "name": "ClusterId", "type": "string", "versions": "1+", "nullableVersions": "1+", "default": "null",
        "about": "The clusterId if known. This is used to validate request against the expected cluster." },
      { "name": "ConsistencyToken", "type": "int64", "versions": "1+", "default": "-1",
        "about": "The latest consistency token seen by the client." }
    ]}
  ]
}

package org.apache.kafka.clients;

/**
 * Object for creating Kafka clients with a shared consistency.
 *
 * This object allows the user to create Admin clients, Producer clients and Consumer clients with
 * a shared consistency.
 *
 * For example, if you would like to create an Admin client to create ACLs and a topic, and have
 * the producer and consumer to see a consistent view of the cluster metadata then use the same
 * factory to create all of the associated clients.
 *
 * This object implements three important menthods. The method {@code admin} can be used to create
 * Admin clients. The method {@code producer} can be used to create Producer clients. The
 * method {@code consumer} can be used to create Consumer clients.
 */
public final class Factory {
    private final ConsistencyContextStore store;

    /**
     * Creates a Factory object.
     *
     * @param store the store for storing the latest consistency context
     */
    Factory(ConsistencyContextStore store) {
        this.store = store;
    }

    /**
     * Creates an Admin client.
     *
     * @param config the admin client configuration
     */
    public Admin admin(Map<String, Object> config) {
        ...
    }

    /**
     * Creates a Producer client.
     *
     * @param config the producer configuration
     * @param keySerializer the serializer for the key
     * @param ValueSerializer the serializer for the value
     */
    public <K, V> Producer<K, V> producer(
        Map<String, Object> config,
        Serializer<K> keySerializer,
        Serializer<V> valueSerializer
    ) {
        ...
    }

    /**
     * Creates a Consumer clients.
     *
     * @param config the consumer configuration
     * @param keyDeserializer the deserializer for the key
     * @param valueDeserializer the deserializer for the value
     */
    public <K, V> Consumer<K, V> consumer(
        Map<String, Object> config,
        Deserializer<K> keyDeserializer,
        Deserializer<V> valueDeserializer
    ) {
        ...
    }
}

package org.apache.kafka.clients;

/**
 * An object that can be used to read and write the latest metadata consistency context for a Kafka
 * cluster.
 *
 * A {@code ConsistencyContextStore} exposes two operations. The method {@code storeLatest} can be
 * used to store the latest metadata consistency context. The method {@code read} can be used to
 * read the latest metadata consistent context.
 */
public interface ConsistencyContextStore {
    /**
     * Method for storing the latest consistency context.
     *
     * Overrides the stored metadata {@code ConsistencyContext} if the provided
     * {@code consistencyContext} is more up to date than the stored metadata consistency context.
     * Otherwise the already stored metadata {@code ConsistencyContext} is kept.
     *
     * Returns {@code true} if the stored metadata consistency context was updated. Otherwise, it
     * returns {@code false}.
     *
     * @param consistencyContext the new consistency context to attempt store
     * @return true the consistency context was updated, otherwise returns false
     * @throws IllegalArgumentException if the cluster id do not match
     */
    boolean storeLatest(ConsistencyContext consistencyContext);

    /**
     * Method for reading the latest consistency context.
     *
     * @return the currently stored metadata consistency context
     */
    ConsistencyContext read();

    /**
     * Creates and returns an {@code ConsistencyContextStore} object with
     * {@code consistencyContext} as the starting value.
     *
     * @param consistencyContext the initial value for the consisteny context store
     * @return a memory backed consistency context store
     */
    public static ConsistencyContextStore of(ConsistencyContext consistencyContext) {
        return MemoryConsistencyContextStore.of(consistencyContext);
    }

    /**
     * Creates and returns an {@code ConsistencyContextStore} object with no initial value.
     *
     * @return a memory backed consistency context store
     */
    public static ConsistencyContextStore empty() {
        return MemoryConsistencyContextStore.of(ConsistencyContext.unknown());
    }

    /**
     * Legacy metadata consistency context.
     *
     * This metadata consistency context store achieves eventual consistency context by always
     * returning the empty consistency context.
     *
     * @return a consistency context store that always the empty consistency context
     */
    public static ConsistencyContextStore eventuallyConsistent() {
        return LegacyConsistencyContextStore.singleton();
    }
}

package org.apache.kafka.common.metadata;

/**
 * An object that represent the metadata consistency context for a given Kafka cluster.
 *
 * A {@code ConsistencyContext} exposes two operations. The method {@code later} compares
 * two consistency context and return the more up to date context. The method {@code isUnknown}
 * returns true if the object represents an unknown metadata consistency context.
 */
public interface ConsistencyContext extends Serializable {
    /**
     * Compares two consistency context and returns the more up to date consistency context.
     *
     * The returned consistency context will have the greater of the two offsets. If the cluster
     * id do not match an {@code IllegalArgumentException} is thrown.
     *
     * @param other the consistency context to compare
     * @return the more up to date consistency context
     * @throws IllegalArgumentException if neither consistency context is empty and the cluster id
     *     do not match
     */
    public ConsistencyContext later(ConsistencyContext other);

    /**
     * Returns true is the consistency context is unknown.
     */
    public boolean isUnknown();

    /**
     * Returns the empty consistency context.
     *
     * This is the default consistency context when the value is unknown.
     */
    public static ConsistencyContext unknown() {
        return MetadataConsistencyContext.unknown();
    }

    /**
     * Returns a consistency context describing the given cluster id and offset.
     *
     * @param clusterId the cluster id
     * @param offset the metadata offset
     * @return an consistency context representing the cluster id and offset
     * @throws IllegalArgumentException if offset is negative
     * @throws NullPointerException if the cluster id is null
     */
    public static ConsistencyContext of(String clusterId, long offset) {
        return MetadataConsistencyContext.of(clusterId, offset);
    }
}