...

In explicit mode, polling works the same, but the application must acknowledge each record explicitly using one of the AcknowledgementType values. It is here that RENEW comes into play, should the application need it.

Client side changes

If record processing might take time:

• The application should use the new acknowledgement type AcknowledgementType.RENEW as an argument to acknowledge(ConsumerRecord, AcknowledgementType) 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.

...

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, we will also update the ShareConsumer.acknowledge(ConsumerRecord, AcknowledgementType) to throw a new exception Errors.UNSUPPORTED_ACKNOWLEDGEMENT_TYPE. This will be handled completely on the client side where the share consumer will use the RPC API version to determine acknowledgement type validity.

When we RENEW a record, the record state for the same maintained by the SharePartition does not change. This means that on subsequent polls by the same member, the record will not be returned by the broker until the lock timeout expires. Due to this, the application might not get any update about the aforementioned record. To remedy this, we propose buffering any RENEW acknowledged records on the share consumer and returning them on subsequent polls until they are re-delivered by the broker at which point the buffer entry can be cleared.

Broker side changes

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 data from the broker side on ShareFetch requests containing RENEW acknowledgements. That way we are guaranteed that the poll()  completes timely. We assume the application uses appropriate concurrency constructs to process records in separate threads.

Compatibility, Deprecation, and Migration Plan

• The existing functionality is not modified. Applications using share consumers in explicit mode will get a new capability of renewing records.
• We are increasing the version number of ShareFetch and ShareAcknowledge requests. If a broker running the old version of the code receives the RENEW  ack type in the request, it should consider this an error and return the appropriate error code to the consumer (Errors.INVALID_REQUEST at the time of writing).
• Though the broker with old and new code can handle RENEW (return error vs renew lock), the share consumer should be smart enough to figure out based on the broker API versions whether to send the RENEW ack or not (point 3 in aforementioned check client side changes section).

Test Plan

• We will be adding new unit tests mainly in core/src/test/java/kafka/server/share/SharePartitionTest.java to verify batch and offset level renewal as well as mix acknowledgement type handling.
• New integration tests will be added in clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java  to verify mix and renew acknowledgement type handling.