KAFKA-19797: Added deliveryCompleteCount to writeState, ShareUpdate and ShareSnapshot schemas #20797
Conversation
…nd ShareSnapshot schemas
github-actions
bot
added
triage
There was a problem hiding this comment.
Thanks for the PR, LGTM! I will wait for @AndrewJSchofield's review on this.
| "type": "request", | ||
| "listeners": ["broker"], | ||
| "name": "WriteShareGroupStateRequest", | ||
| "validVersions": "0", |
There was a problem hiding this comment.
Please can we have a comment to describe the versions. Here's my suggestion:
// Version 0 is the initial version (KIP-932).
//
// Version 1 introduces DeliveryCompleteCount (KIP-1226).
| "apiKey": 85, | ||
| "type": "response", | ||
| "name": "WriteShareGroupStateResponse", | ||
| "validVersions": "0", |
There was a problem hiding this comment.
Again, a version comment such as:
// Version 0 is the initial version (KIP-932).
//
// Version 1 introduces DeliveryCompleteCount in the request (KIP-1226).
| .setTopicsData(List.of(new TopicData<>(topicIdPartition.topicId(), | ||
| List.of(PartitionFactory.newPartitionStateBatchData( | ||
| topicIdPartition.partition(), stateEpoch, startOffset(), leaderEpoch, stateBatches)))) | ||
| topicIdPartition.partition(), stateEpoch, startOffset(), inFlightTerminalRecords(), leaderEpoch, stateBatches)))) |
There was a problem hiding this comment.
I know that inFlightTerminalRecords was the previous name before the KIP was approved, but it's changed now. Please can we expunge the old name for the sake of the future maintainers.
| verifyNoErr: Boolean = true, authorizer: Authorizer = null, | ||
| writeStateResult: util.List[WriteShareGroupStateResponseData.WriteStateResult]): WriteShareGroupStateResponse = { | ||
| val requestChannelRequest = buildRequest(new WriteShareGroupStateRequest.Builder(requestData).build()) | ||
| val requestChannelRequest = buildRequest(new WriteShareGroupStateRequest.Builder(requestData).build(0)) |
There was a problem hiding this comment.
Why are you hardcoding version 0 here?
There was a problem hiding this comment.
Thanks for the review. That was a typo, will remove the 0.
|
|
||
| public static ShareGroupOffset fromRequest(InitializeShareGroupStateRequestData.PartitionData data, int snapshotEpoch, long timestamp) { | ||
| // Since initialization changes the start offset, and hence the in flight state is forgotten, the end offset is set | ||
| // to be the same as the start offset, and the in flight record count is set to 0. |
There was a problem hiding this comment.
This comment seems inaccurate. I don't understand how the in-flight record count is set to zero here. I do think that the delivery complete count is unknown at this point and is set to the uninitialised value.
| "about": "The leader epoch of the share-partition." }, | ||
| { "name": "StartOffset", "type": "int64", "versions": "0+", | ||
| "about": "The share-partition start offset." }, | ||
| { "name": "DeliveryCompleteCount", "type": "int32", "versions": "0+", "taggedVersions": "0+", "tag": 0, "default": "-1", |
There was a problem hiding this comment.
If you look at ConsumerGroupMetadataValue, you'll see an example of how a new tagged field was commented. I suggest a similar thing here such as
// DeliveryCompleteCount was added in Apache Kafka 4.2 (KIP-1226).
| "about": "The leader epoch of the share-partition." }, | ||
| { "name": "StartOffset", "type": "int64", "versions": "0+", | ||
| "about": "The share-partition start offset, or -1 if the start offset is not being updated." }, | ||
| { "name": "DeliveryCompleteCount", "type": "int32", "versions": "0+", "taggedVersions": "0+", "tag": 0, "default": "-1", |
3 participants
This PR is part of
KIP-1226.
It introduces a new field called
deliveryCompleteCountinwriteShareGroupState RPC and
ShareUpdateandShareSnapshot. Theshare partition now maintains
inFlightTerminalCount(introducedhere) and it uses this
exact value to pass to the writeState request for
deliveryCompleteCount.