Versions Compared

Old Version 2

changes.mady.by.user Unknown User (jsancio)

Saved on

compared with

New Version 3

changes.mady.by.user Unknown User (jsancio)

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Since some read operations are sent the least loaded node, there is no guarantee that future requests from the same client will see a consistent state that is more recent (larger log offset) than previously seen state. This is because there is no guarantee that future read operations will be sent to a node that contain previously seen log offsets.

Proposed Changes

Describe the new thing you want to do in appropriate detail. This may be fairly extensive and have large subsections of its own. Or it may be a few sentences. Use judgement based on the scope of the change.

Public Interfaces

Remote Produce Calls

...

Server

Kafka public API will be extended by allowing clients to provide the latest metadata offset known by the client. This information will be encoded in the RequestHeader using the cluster metadata offset. When a server handles a request it will check that its own cluster metadata offset is as up to date as the client's. If the server is behind the client it may wait up to the request timeout for its own cluster metadata to be as update as the client's or it may return an INCONSISTENT_METADATA error.

After handling a request, Kafka servers will include their latest cluster metadata offset in the ResponseHeader.

Clients

TODO: explain this...

Public Interfaces

Remote Produce Calls

Error

Kafka will add a new retrievable error named INCONSISTENT_METADATA. This error will be return by the Kafka servers if their cluster metadata version is not as up to date as the client provided cluster metadata version. This error is not fatal and it is retrievable. The client should retry the operation if they receive an INCONSISTENT_METADATA error.

RequestHeader

Code Block
{
  "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." }
    ]}
  ]
}

ResponseHeader

Code Block
{
  "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." }
    ]}
  ]
}

Handling

TODO

Sending

TODO

Clients

Factory

Code Block
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
    ) {
        ...
    }
}

...

Code Block
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);

    /** empty and the cluster id
     * Returns true is the consistencydo context is unknown.not match
     */
    public booleanConsistencyContext isUnknownlater(ConsistencyContext other);

    /**
     * Returns the empty consistency context.
     *
     * This true is the default consistency context when the value is unknown.
     */
    public staticboolean ConsistencyContext unknownisUnknown() {;

    /**
    return MetadataConsistencyContext.unknown();
    }

 * Returns the empty consistency context.
     /**
     * This is Returnsthe adefault consistency context describingwhen the givenvalue cluster id and offsetis unknown.
     */
    public  * @param clusterId the cluster id
     * @param offset the metadata offsetstatic ConsistencyContext unknown() {
        return MetadataConsistencyContext.unknown();
    }

    /**
     * @returnReturns ana consistency context representingdescribing the given cluster id and offset.
     * @throws IllegalArgumentException if offset is negative
     * @throws@param NullPointerException ifclusterId the cluster id is null
     */
 @param offset the public static ConsistencyContext of(String clusterId, long offset) {
        return MetadataConsistencyContext.of(clusterId, offset);
    }
}

Briefly list any new interfaces that will be introduced as part of this proposal or any existing interfaces that will be removed or changed. The purpose of this section is to concisely call out the public contract that will come along with this feature.

A public interface is any change to the following:

...

Binary log format

...

The network protocol and api behavior

...

Any class in the public packages under clientsConfiguration, especially client configuration

  • org/apache/kafka/common/serialization

  • org/apache/kafka/common

  • org/apache/kafka/common/errors

  • org/apache/kafka/clients/producer

  • org/apache/kafka/clients/consumer (eventually, once stable)

...

Monitoring

...

Command line tools and arguments

...

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);
    }
}

Compatibility, Deprecation, and Migration Plan

...