Page History

...

• 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 255field 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": "0+", "about": "The minimum bytes to accumulate in the response." }, { "name": "MaxBytes", "type": "int32", "versions": "0+", "default": "0x7fffffff", "about": "The maximum bytes to fetch. See KIP-74 for cases where this limit may not be honored." }, { "name": "MaxRecords", "type": "int32", "versions": "1+", "about": "The maximum number of records to fetch. This limit can be exceeded for alignment of batch boundaries." }, { "name": "BatchSize", "type": "int32", "versions": "1+", "about": "The optimal number of records for batches of acquired records and acknowledgements." }, { "name": "ShareAcquireModeIsRenewAck", "type": "int8bool", "versions": "2+", "default": false, "about": "TheWhether acquirerenew modetype toacknowledgement controlpresent thein fetch behavior: 0 - batch-optimized, 1 - record-limit, 255 - renew-ack-mode" }, // From KIP-1206AcknowledgementBatches." }, // Version 2 supports RENEW ack type and this field serves as an indicator (KIP-1222) { "name": "Topics", "type": "[]FetchTopic", "versions": "0+", "about": "The topics to fetch.", "fields": [ { "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID.", "mapKey": true }, { "name": "Partitions", "type": "[]FetchPartition", "versions": "0+", "about": "The partitions to fetch.", "fields": [ { "name": "PartitionIndex", "type": "int32", "versions": "0+", "mapKey": true, "about": "The partition 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." }, { "name": "AcknowledgementBatches", "type": "[]AcknowledgementBatch", "versions": "0+", "about": "Record batches to acknowledge.", "fields": [ { "name": "FirstOffset", "type": "int64", "versions": "0+", "about": "First offset of batch of records to acknowledge."}, { "name": "LastOffset", "type": "int64", "versions": "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)         ]} ]} ]}, { "name": "ForgottenTopicsData", "type": "[]ForgottenTopic", "versions": "0+", "about": "The partitions to remove from this share session.", "fields": [ { "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."}, { "name": "Partitions", "type": "[]int32", "versions": "0+", "about": "The partitions indexes 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+", "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": "TopicsIsRenewAck", "type": "[]AcknowledgeTopicbool", "versions": "02+", "default": false, "about": "TheWhether renew topicstype containingacknowledgement recordspresent toin acknowledgeAcknowledgementBatches." }, "fields": [ { "name": "// 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": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID.", "mapKey": true }, { "name": "Partitions", "type": "[]AcknowledgePartition", "versions": "0+", "about": "The partitions containing records to acknowledge.", "fields": [ { "name": "PartitionIndex", "type": "int32", "versions": "0+", "mapKey": true, "about": "The partition index." }, { "name": "AcknowledgementBatches", "type": "[]AcknowledgementBatch", "versions": "0+", "about": "Record batches to acknowledge.", "fields": [ { "name": "FirstOffset", "type": "int64", "versions": "0+", "about": "First offset of batch of records to acknowledge." }, { "name": "LastOffset", "type": "int64", "versions": "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 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();

   

...

• 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 ShareAcquireMode IsRenewAck set to 255 and maxWaitMs, maxBytes and maxRecords to 0 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 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. In order to make it clear when a ShareFetch request request is renewing rather than acquiring records, a new value of ShareAcquireMode of RENEW (255) is usedIsRenewAck will be set to true. This also implies that we will not honour the maxWaitMsMaxWaitMs, MinBytes, maxBytes MaxBytesand maxRecords MaxRecords in the ShareFetch request containing ShareAcquireMode IsRenewAck value of 255 true. Hence, we have mentioned in the aforementioned section, that the share consumer should set these fields to 0. By looking at the value of the top level field IsRenewAck, the broker could process the request a bit more optimally.

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.

...