Table of Contents

Status

Current state: Under discussionAccepted

Discussion thread: here

Voting thread: here

JIRA:

Jira

server	ASF JIRA
serverId	5aa69414-a9e9-3523-82ec-879b028fb15b
key	KAFKA-19742

...

Code Block

public enum AcknowledgeType {
    /** The record was consumed successfully. */
    ACCEPT((byte) 1),
    /** The record was not consumed successfully. Release it for another delivery attempt. */
    RELEASE((byte) 2),
    /** The record was not consumed successfully. Reject it and do not release it for another delivery attempt. */
    REJECT((byte) 3),
    /** Consumer needs more time to process the record. Renew the lease. */
    RENEW((byte) 4);	// New entry per KIP-1222
    ...
}

There The acknowledgement type will be an addition used in the org.apache.kafka.clients.consumer.ShareAcquireMode enum defined in KIP-1206. existing ShareFetch and ShareAcknowledge RPCs. Hence, we must add new versions for them. Additionally, we will add a field IsRenewAck defaulting to false to both the RPCs. This field will help in optimising the broker side implementation for processing renew acknowledgements.

clients/src/main/resources/common/message/ShareFetchRequest.json

Code Block

// ShareFetch
{
  "apiKey": 78,
  "type": "request",
  "listeners": ["broker"],
  "name": "ShareFetchRequest",
  // Version 0 was used for early access of KIP-932 in Apache Kafka 4.0 but removed in Apacke Kafka 4.1.
  //
  // Version 1 is the initial stable version (KIP-932).
  //
  // Version 2 supports RENEW ack type
  "validVersions": "1-2",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null", "entityType": "groupId",
      "about": "The group identifier." },
    { "name": "MemberId", "type": "string", "versions": "0+", "nullableVersions": "0+",
      "about": "The member ID." },
    { "name": "ShareSessionEpoch", "type": "int32", "versions": "0+",
      "about": "The current share session epoch: 0 to open a share session; -1 to close it; otherwise increments for consecutive requests." },
    { "name": "MaxWaitMs", "type": "int32", "versions": "0+",
      "about": "The maximum time in milliseconds to wait for the response." },
    { "name": "MinBytes", "type": "int32", "versions

Code Block

public enum ShareAcquireMode {
    BATCH_OPTIMIZED("batch_optimized"),
    RECORD_LIMIT("record_limit"),
	RENEW_ACK_MODE("renew_ack_mode"); // Added as part for KIP-1222
     
    public final String name;
     
    ShareAcquireMode(final String name) {
        this.name = name;
    }
 
    /**
     * Case-insensitive acquire mode lookup by string name.
     */
    public static ShareAcquireMode of(final String name) {
        return ShareAcquireMode.valueOf(name.toUpperCase(Locale.ROOT));
    }
}

The acknowledgement type will be used in the existing ShareFetch and ShareAcknowledge RPCs. Hence, we must add new versions for them. Additionally, we will add a new value for the field shareAcquireMode called renew-ack-mode with value id value 255. This field will help in optimising the broker side implementation for processing renew acknowledgements.

clients/src/main/resources/common/message/ShareFetchRequest.json

Code Block

// ShareFetch
{
  "apiKey": 78,
  "type": "request",
  "listeners": ["broker"],
  "name": "ShareFetchRequest",
  // Version 0 was used for early access of KIP-932 in Apache Kafka 4.0 but removed in Apacke Kafka 4.1.
  //
  // Version 1 is the initial stable version (KIP-932).
  //
  // Version 2 supports RENEW ack type
  "validVersions": "1-2",
  "flexibleVersions": "0+",
      "fieldsabout": [ "The minimum bytes to accumulate in the response." },
    { "name": "GroupIdMaxBytes", "type": "stringint32", "versions": "0+", "nullableVersions": "0+", "default": "null", "entityType": "groupId0x7fffffff",
      "about": "The group identifiermaximum bytes to fetch. See KIP-74 for cases where this limit may not be honored." },
    { "name": "MemberIdMaxRecords", "type": "stringint32", "versions": "0+", "nullableVersions": "01+",
      "about": "The member ID maximum number of records to fetch. This limit can be exceeded for alignment of batch boundaries." },
    { "name": "ShareSessionEpochBatchSize", "type": "int32", "versions": "01+",
      "about": "The currentoptimal sharenumber sessionof epoch:records 0for tobatches openof aacquired sharerecords session; -1 to close it; otherwise increments for consecutive requestsand acknowledgements." },
    { "name": "MaxWaitMsIsRenewAck", "type": "int32bool", "versions": "02+", "default": false,
      "about": "TheWhether maximumrenew timetype inacknowledgement millisecondspresent to wait for the responsein AcknowledgementBatches." }, // Version 2 supports RENEW ack type and this field serves as an indicator (KIP-1222)
    { "name": "MinBytesTopics", "type": "int32[]FetchTopic", "versions": "0+",
      "about": "The minimum bytestopics to accumulate in the response." },
fetch.", "fields": [
      { "name": "MaxBytesTopicId", "type": "int32uuid", "versions": "0+", "defaultabout": "0x7fffffff",
The unique topic  ID.",  "aboutmapKey": "The maximum bytes to fetch. See KIP-74 for cases where this limit may not be honored." true },
      { "name": "MaxRecordsPartitions", "type": "int32[]FetchPartition", "versions": "10+",
        "about": "The maximum number of recordspartitions to fetch. This limit can be exceeded for alignment of batch boundaries." },
    {", "fields": [
        { "name": "BatchSizePartitionIndex", "type": "int32", "versions": "10+", "mapKey":  true,
          "about": "The optimal number of records for batches of acquired records and acknowledgementspartition index." },
        { "name": "ShareAcquireModePartitionMaxBytes", "type": "int8int32", "versions": "2+0",
          "about": "The acquiremaximum modebytes to controlfetch thefrom fetchthis behavior:partition. 0 when only acknowledgement with no fetching is required. See KIP- batch-optimized, 1 - record-limit, 255 - renew-ack-mode74 for cases where this limit may not be honored." },
  // From KIP-1206
    { "name": "TopicsAcknowledgementBatches", "type": "[]FetchTopicAcknowledgementBatch", "versions": "0+",
          "about": "TheRecord topicsbatches to fetchacknowledge.", "fields": [
          { "name": "TopicIdFirstOffset", "type": "uuidint64", "versions": "0+",
            "about": "The unique topic ID.", "mapKey":  true },
First offset of batch of records to acknowledge."},
          { "name": "PartitionsLastOffset", "type": "[]FetchPartitionint64", "versions": "0+",
            "about": "The partitionsLast offset (inclusive) of batch of records to fetchacknowledge."},
 "fields": [
        { "name": "PartitionIndexAcknowledgeTypes", "type": "int32[]int8", "versions": "0+",
 "mapKey":  true,
          "about": "TheArray partitionof index." },
        { "name": "PartitionMaxBytes", "type": "int32", "versions": "0",
          "about": "The maximum bytes to fetch from this partition. 0 when only acknowledgement with no fetching is required. See KIP-74 for cases where this limit may not be honored." },
        acknowledge types - 0:Gap,1:Accept,2:Release,3:Reject,4:Renew."} // Version 2 supports RENEW ack type (KIP-1222)
         ]}
      ]}
    ]},
    { "name": "AcknowledgementBatchesForgottenTopicsData", "type": "[]AcknowledgementBatchForgottenTopic", "versions": "0+",
          "about"about": "RecordThe batchespartitions to acknowledge remove from this share session.", "fields": [
          { "name": "FirstOffsetTopicId", "type": "int64uuid", "versions": "0+",
            "about": "FirstThe offsetunique of batch of records to acknowledgetopic ID."},
          { "name": "LastOffsetPartitions", "type": "int64[]int32", "versions": "0+",
            "about": "LastThe offsetpartitions (inclusive)indexes of batch of records to acknowledgeto forget." },
    ]}
  ]
}

clients/src/main/resources/common/message/ShareAcknowledgeRequest.json

Code Block

// ShareAcknowledge
{
  "apiKey": 79,
 { "nametype": "AcknowledgeTypesrequest",
  "typelisteners": ["[broker"]int8",
  "versionsname": "0+ShareAcknowledgeRequest",
  // Version 0 was used for early access of KIP-932 in Apache  "about": "Array of acknowledge types - 0:Gap,1:Accept,2:Release,3:Reject,4:Renew."}Kafka 4.0 but removed in Apache Kafka 4.1.
  //
  // Version 1 2is supportsthe RENEWinitial ackstable typeversion (KIP-1222932).
    //
     ]}
      ]}
    ]},// Version 2 will have RENEW ack type
  "validVersions": "1-2",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ForgottenTopicsDataGroupId", "type": "[]ForgottenTopicstring", "versions": "0+",
 "nullableVersions": "0+", "default": "null",  "aboutentityType": "groupId"The,
 partitions to remove from this share session.", "fields": [
  "about": "The group identifier." },
    { "name": "TopicIdMemberId", "type": "uuidstring", "versions": "0+", "nullableVersions": "0+",
      "about": "The unique topicmember ID." },
      { "name": "PartitionsShareSessionEpoch", "type": "[]int32", "versions": "0+",
        "about": "The partitionscurrent indexesshare to forget." }
    ]}
  ]
}

clients/src/main/resources/common/message/ShareAcknowledgeRequest.json

Code Block

// ShareAcknowledge
{
  "apiKey": 79,
  "type": "request",
  "listeners": ["broker"],
  "name": "ShareAcknowledgeRequest",
  // Version 0 was used for early access of KIP-932 in Apache Kafka 4.0 but removed in Apacke Kafka 4.1.
  //
  // Version 1 is the initial stable version (KIP-932).
  //
  // Version 2 will have RENEW ack type
  "validVersions": "1-2",
  "flexibleVersions": "0+",
  session epoch: 0 to open a share session; -1 to close it; otherwise increments for consecutive requests." },
    { "name": "IsRenewAck", "type": "bool", "versions": "2+", "default": false,
      "about": "Whether renew type acknowledgement present in AcknowledgementBatches." }, // Version 2 supports RENEW ack type and this field serves as an indicator (KIP-1222)
    { "name": "Topics", "type": "[]AcknowledgeTopic", "versions": "0+",
      "about": "The topics containing records to acknowledge.", "fields": [
      { "name": "GroupIdTopicId", "type": "stringuuid", "versions": "0+", "nullableVersionsabout": "0+", "default": "null"The unique topic ID.", "entityTypemapKey": true "groupId"},
      "about": "The group identifier." },
    { "name": "MemberIdPartitions", "type": "string[]AcknowledgePartition", "versions": "0+",
 "nullableVersions": "0+",
      "about": "The member ID." },
 partitions containing records to acknowledge.", "fields": [
        { "name": "ShareSessionEpochPartitionIndex", "type": "int32", "versions": "0+",
      "aboutmapKey": "The current share session epoch: 0 to open a share session; -1 to close it; otherwise increments for consecutive requeststrue,
          "about": "The partition index." },
        { "name": "TopicsAcknowledgementBatches", "type": "[]AcknowledgeTopicAcknowledgementBatch", "versions": "0+",
          "about": "TheRecord topics containingbatches records to acknowledge.", "fields": [
          { "name": "TopicIdFirstOffset", "type": "uuidint64", "versions": "0+", "
            "about": "The unique topic ID.", "mapKey": true },
First offset of batch of records to acknowledge." },
          { "name": "PartitionsLastOffset", "type": "[]AcknowledgePartitionint64", "versions": "0+",
            "about": "The partitions containingLast offset (inclusive) of batch of records to acknowledge." },
 "fields": [
        { "name": "PartitionIndexAcknowledgeTypes", "type": "int32[]int8", "versions": "0+", "mapKey": true,

            "about": "TheArray partitionof index." },
        { "name": "AcknowledgementBatches", "type": "[]AcknowledgementBatch", "versions": "0+",
          "about": "Record batches to acknowledge.", "fields": [
          { "name": "FirstOffset",acknowledge types - 0:Gap,1:Accept,2:Release,3:Reject,4:Renew" } // Version 2 supports RENEW ack type (KIP-1222)
         ]}
      ]}
    ]}
  ]
}

clients/src/main/resources/common/message/ShareAcknowledgeResponse.json (Response will contain a new field AcquisitionLockTimeoutMs from version 2 onwards).

Code Block

{
  "apiKey": 79,
  "type": "int64response",
  "versionsname": "0+ShareAcknowledgeResponse",
  // Version 0 was used for early access of KIP-932 in "about": "First offset of batch of records to acknowledge." },
          { "name": "LastOffset", "type": "int64", "versionsApache Kafka 4.0 but removed in Apache Kafka 4.1.
  //
  // Version 1 is the initial stable version (KIP-932).
  //
  // Version 2 introduces Renew acknowledgements (KIP-1222).
  "validVersions": "1-2",
  "flexibleVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)
  "about": "Last offset (inclusive) of batch of records to acknowledge." },
          { "name": "AcknowledgeTypes", "type": "[]int8", "versions": "0+",
            "about": "Array of acknowledge types - 0:Gap,1:Accept,2:Release,3:Reject,4:Renew" } // Version 2 supports RENEW ack type (KIP-1222)
         ]}
      ]}
    ]}
  ]
}

New method exposed in clients/src/main/java/org/apache/kafka/clients/consumer/ShareConsumer.java interface to help the application determine RENEW interval.

Code Block/** Returns the acquisition lock timeout value

// - TOPIC_AUTHORIZATION_FAILED (version 0+)
  // - UNKNOWN_TOPIC_OR_PARTITION (version 0+)
  // - SHARE_SESSION_NOT_FOUND (version 0+)
  // - INVALID_SHARE_SESSION_EPOCH (version 0+)
  // - NOT_LEADER_OR_FOLLOWER (version 0+)
  // - UNKNOWN_TOPIC_ID (version 0+)
  // - INVALID_RECORD_STATE (version 0+)
  // - KAFKA_STORAGE_ERROR (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - UNKNOWN_SERVER_ERROR (version 0+)
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the

last

request

set

was

of

throttled

records

due

fetched

to

from

a

the brokers. This is an optional field as unless an application poll() results in a ShareFetch to a broker, this value cannot be determined. */ public Optional<Integer> acquisitionLockTimeoutMs();

Proposed Changes

quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The top level response error code." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." },
    { "name": "AcquisitionLockTimeoutMs", "type": "int32", "versions": "2+",
      "about": "The time in milliseconds for which the acquired records are locked." },
    { "name": "Responses", "type": "[]ShareAcknowledgeTopicResponse", "versions": "0+",
      "about": "The response topics.", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+", "mapKey": true,
        "about": "The unique topic ID." },
      { "name": "Partitions", "type": "[]PartitionData", "versions": "0+",
        "about": "The topic partitions.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
          "about": "The partition index." },
        { "name": "ErrorCode", "type": "int16", "versions": "0+",
          "about": "The error code, or 0 if there was no error." },
        { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
          "about": "The error message, or null if there was no error." },
        { "name": "CurrentLeader", "type": "LeaderIdAndEpoch", "versions": "0+",
          "about": "The current leader of the partition.", "fields": [
          { "name": "LeaderId", "type": "int32", "versions": "0+",
            "about": "The ID of the current leader or -1 if the leader is unknown." },
          { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
            "about": "The latest known leader epoch." }
        ]}
      ]}
    ]},
    { "name": "NodeEndpoints", "type": "[]NodeEndpoint", "versions": "0+",
      "about": "Endpoints for all current leaders enumerated in PartitionData with error NOT_LEADER_OR_FOLLOWER.", "fields": [
      { "name": "NodeId", "type": "int32", "versions": "0+",
        "mapKey": true, "entityType": "brokerId", "about": "The ID of the associated node." },
      { "name": "Host", "type": "string", "versions": "0+",
        "about": "The node's hostname." },
      { "name": "Port", "type": "int32", "versions": "0+",
        "about": "The node's port." },
      { "name": "Rack", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "The rack of the node, or null if it has not been assigned to a rack." }
    ]}
  ]
}

New method exposed in clients/src/main/java/org/apache/kafka/clients/consumer/ShareConsumer.java interface to help the application determine RENEW interval.

Code Block

/**
 Returns the acquisition lock timeout value in milliseconds for the last set of records fetched from the brokers.
 This is an optional field as unless an application poll() results in a ShareFetch to a broker, this value cannot be
 determined.
*/
public Optional<Integer> acquisitionLockTimeoutMs();

An additional tag for the metric RecordAcknowledgementsPerSec, defined in KIP-1103, will get introduced as a side effect of introducing the RENEW ack type. The new definition of the metric is:

Rationale Metric Name Type Group Tags Description JMX Bean

Tracks the rate of records acknowledged per acknowledgement type.

RecordAcknowledgementsPerSec

Meter

ShareGroupMetrics

ackType:{Accept|Release|Reject|Renew}

The rate per second of records acknowledged per acknowledgement type.

kafka.server:type=ShareGroupMetrics,name=RecordAcknowledgementsPerSec,ackType={Accept|Release|Reject|Renew}

Proposed Changes

KIP-932 added two new RPCs for record fetching and acknowledging: ShareFetch and ShareAcknowledge. From the share consumer, these KIP-932 added two new RPCs for record fetching and acknowledging: ShareFetch and ShareAcknowledge. From the share consumer, these RPCs are called when an application invokes poll(Duration)and commitSync(Duration)/commitAsync().

...

The application should use the new acknowledgement type AcknowledgeType.RENEW as an argument to acknowledge(ConsumerRecord, AcknowledgeType) for that record.
It can then call either commitSync()/commitAsync() or poll() to send the acknowledgement to the server. Furthermore, the application could also leverage the new ShareConsumer.acquisitionLockTimeoutMs() method to decide on the timeframe in which to make the RENEW call. If the subsequent call is poll(), the share consumer will create a ShareFetch request with IsRenewAck set to true and MaxWaitMs, MinBytes, MaxBytesandMaxRecordsto 0. If instead of poll(), commitSync()/commitAsync() are called, IsRenewAck will be set appropriately as well.

There is a case where an application tries to send a RENEW acknowledgement to an old broker which does not support the new type. To remedy this, on the next call to poll/commitSync/commitAsync, the share consumer will verify if the ack type is supported and notify the application using the appropriate callback handler. The share consumer will return the exception UnsupportedVersionException (Errors.UNSUPPORTED_VERSION (35)) as response to a RENEW request meant for an unsupported broker. This will be handled completely on the client side where the share consumer will use the API version RPC to determine acknowledgement type validity.

...

On the broker side, receiving a RENEW acknowledgement for a specific batch or offset will cancel the existing acquisition lock timeout task and start a new one with the same timeout value as group.share.record.lock.duration.ms. There is one caveat here. If the application causes a ShareFetch RPC to be sent (poll() call) on which RENEW acknowledgements are piggybacked, it could happen that the renewed acquisition lock again times out before the poll() completes. To get around this, we will not return any new records from the broker side on ShareFetch requests containing RENEW acknowledgementscontaining RENEW acknowledgements (indicated by the IsRenewAck field). That way we are guaranteed that the poll() completes timely and we can return the response of acknowledgements immediately to the application. This also implies that we will not honour the maxWaitMs, maxBytes and maxRecords in the ShareFetch request containing RENEW ack type. Hence, these 3 fields must be set to 0 for requests containing RENEW acks.that the poll() completes timely and we can return the response of acknowledgements immediately to the application. A true value of IsRenewAck makes it clear that a ShareFetch request is renewing rather than acquiring records. If IsRenewAck is set to true, MaxWaitMs, MinBytes, MaxBytes and MaxRecords must be set to 0. Also, by looking at the value of IsRenewAck, the broker could process the request a bit more optimally.

If a share consumer sends a ShareAcknowledge request for renewal, the response should provide the value of the acquisition lock timeout. This is because the timeout is a configurable field. It could happen that a record is fetched with a specific timeout, then the timeout value is updated on the broker and then the share consumer issues a ShareAcknowledge RENEW call on the aforementioned record. Now the renewed timeout should be timely communicated to the consumer. The new field AcquisitionLockTimeoutMs in ShareAcknowledge response will fulfil this use caseThe broker will also make use of the ShareAcquireMode to efficiently check if the ShareFetch contains RENEW acknowledgements and this will help in returning the response faster to the share consumer after renewing the acquisition lock.

If the RENEW request received by a broker is invalid and an error is returned, the application should use existing acknowledgement commit callback to listen for the same. Specifically, the application should set a commit response handler callback in ShareConsumer.setAcknowledgementCommitCallback(AcknowledgementCommitCallback) and handle errors, if any. This is the usual mechanism to check commit status in explicit mode and this KIP does not make any change in this code path.

...

Space shortcuts

Child pages

Versions Compared

Old Version 16

New Version Current

Key

Status

Proposed Changes

Proposed Changes

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 16

New Version Current

Key

Status

Proposed Changes

Proposed Changes