...

This KIP proposes a new acknowledgement type, RENEW, for explicit mode. It lets a share consumer request the broker to renew the acquisition lock timeout, thereby extending the current delivery attempt without changing the server's record state.

Public Interfaces

• There will be
new
• an addition in the org/apache/kafka/clients/consumer/AcknowledgeType.java  enum where we will add a new entry RENEW with id 4. 

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
    ...
}

• The acknowledgement type There will be used an addition in the existing org.apache.kafka.clients.consumer.ShareAcquireMode enum defined in KIP-1206. 

  Code Block
  public enum ShareAcquireMode { BATCH_OPTIMIZED("batch_optimized"), RECORD_LIMIT("record_limit"), RENEW_ACK_MODE("renew_ack_mode"); 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,

  ShareFetch and ShareAcknowledge RPCs. Hence, we must add new versions for them.

  • 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+", "versions"nullableVersions": "0+", "about": "The member ID." }, { "name": "ShareSessionEpoch", "type": "0+int32", "nullableVersionsversions": "0+", "about": "The member ID current share session epoch: 0 to open a share session; -1 to close it; otherwise increments for consecutive requests." }, { "name": "ShareSessionEpochMaxWaitMs", "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 "The maximum time in milliseconds to wait for the response." }, { "name": "MaxWaitMsMinBytes", "type": "int32", "versions": "0+", "about": "The maximumminimum timebytes in milliseconds to waitaccumulate forin the response." }, { "name": "MinBytesMaxBytes", "type": "int32", "versions": "0+", "default": "0x7fffffff", "about": "The minimummaximum bytes to accumulate in the response to fetch. See KIP-74 for cases where this limit may not be honored." }, { "name": "MaxBytesMaxRecords", "type": "int32", "versions": "01+", "default": "0x7fffffff", "about": "The maximum number of bytesrecords to fetch. SeeThis KIP-74limit forcan casesbe whereexceeded thisfor limitalignment mayof notbatch be honoredboundaries." }, { "name": "MaxRecordsBatchSize", "type": "int32", "versions": "1+", "about": "The maximumoptimal number of records to fetch. This limit can be exceeded for alignmentbatches of batch boundariesacquired records and acknowledgements." }, { "name": "BatchSizeShareAcquireMode", "type": "int32int8", "versions": "12+", "about": "The optimal number of records for batches of acquired records and acknowledgements." }, acquire mode to control the fetch behavior: 0 - batch-optimized, 1 - record-limit, 255 - renew-ack-mode" }, // From KIP-1206 { "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": "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. */ public Optional<Integer> acquisitionLockTimeoutMs();

   

...

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 as response to a renew RENEW request to meant for an unsupported broker. This will be handled completely on the client side where the share consumer will use the RPC 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. 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 acknowledgements.

The 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.

...