gabriellefu opened a new pull request, #22634:
URL: https://github.com/apache/kafka/pull/22634
# Streams group members can now report TaskOffsets and TaskEndOffsets in
heartbeats
Streams group members can now include cumulative changelog offsets
(`TaskOffsets`) and end offsets (`TaskEndOffsets`) in heartbeat requests.
Previously, the group coordinator rejected these fields with
`INVALID_REQUEST` (`"TaskOffsets/TaskEndOffsets are not supported yet."`). This
change wires the fields through the coordinator so they are:
* persisted in the member's current-assignment record
* surfaced in the `StreamsGroupDescribe` response
## What changed
### 1. Coordinator validation
* Removed the `throwIfNotNull` guards in `GroupCoordinatorService` that
rejected `taskOffsets` and `taskEndOffsets`
### 2. Member model
* Added `taskOffsets` and `taskEndOffsets` (`Map<TaskId, Long>`) to
`StreamsGroupMember`
* Updated the record, `Builder`, `withDefaults`, and `updateWith(...)`
* Store both maps as unmodifiable
### 3. Heartbeat handling
* In `GroupMetadataManager`, convert the request's task offsets into
per-task maps
* Set those maps on the updated member
* If offsets change but reconciliation did not already write a new
current-assignment record, append a record so the updated offsets are replayed
into member state and become visible in describe
### 4. Record schema
* Added tagged fields to `StreamsGroupCurrentMemberAssignmentValue.json`
* `TaskOffsets` (tag 0)
* `TaskEndOffsets` (tag 1)
* Added a shared `TaskOffset` struct containing:
* `SubtopologyId`
* `Partition`
* `Offset`
* Using tagged fields keeps the change backward compatible
### 5. Record and response serialization
* `StreamsCoordinatorRecordHelpers` now writes offsets into the assignment
record
* `StreamsGroupMember.asStreamsGroupDescribeMember` now emits offsets in the
describe response
* Both paths emit deterministically sorted `TaskId` output
## Testing
### End-to-end coverage
*
`GroupMetadataManagerTest.testStreamsGroupHeartbeatPersistsTaskOffsetsAndSurfacesThemInDescribe`
* A member joins
* Sends a follow-up heartbeat with task offsets and end offsets
* The describe response surfaces the reported values
### Unit coverage
* `StreamsGroupMemberTest` now covers:
* builder defaults
* full builder round-trip
* describe-response mapping for the new fields
### Cleanup
* Removed obsolete assertions from `GroupCoordinatorServiceTest` that
expected rejection errors
## Result
Streams members can now report task offset progress in heartbeats, and that
information is durably stored and returned by `StreamsGroupDescribe`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
