wilmerdooley opened a new pull request, #22610:
URL: https://github.com/apache/kafka/pull/22610
## Description
`MirrorCheckpointTask` previously treated the absence of a consumer group
from `idleConsumerGroupsOffset` as evidence that the group was new on the
target cluster and unconditionally synced the converted upstream offsets for
it. In practice this also covered groups that were actively being consumed on
the target (e.g. `STABLE`) but were momentarily not present in the snapshot
because the group transitioned out of `EMPTY` between
`refreshIdleConsumerGroupOffset` calls. The result was that active consumers
could be rewound, or, when the older offset was no longer retained, reset to
`earliest`/`latest` and either re-read or skip messages.
This change tracks the last observed `GroupState` of each target consumer
group alongside the existing offset snapshot, and `syncGroupOffset` now only
syncs offsets for groups reported as `DEAD` (new on the target) or `EMPTY` (and
only when an offset snapshot for them is available). Active groups are skipped,
and the maps are reset at the start of every refresh and in a `finally` block
after syncing so a stale snapshot cannot be applied if the group has since
become active. A transient failure to refresh a group's state or offsets (e.g.
a `listConsumerGroupOffsets` timeout) now leaves the group absent from both
maps, so it is also skipped rather than synced based on stale information. The
diff also adds a `DEAD`-state test and updates the existing tests to populate
the new state map.
Delete this text and replace it with a detailed description of your change.
The
PR title and body will become the squashed commit message.
If you would like to tag individuals, add some commentary, upload images, or
include other supplemental information that should not be part of the
eventual
commit message, please use a separate comment.
If applicable, please include a summary of the testing strategy (including
rationale) for the proposed change. Unit and/or integration tests are
expected
for any behavior change and system tests should be considered for larger
changes.
### Testing strategy
Unit tests in `MirrorCheckpointTaskTest` are updated to populate the new
`targetConsumerGroupStates` map for the existing `testSyncOffset` and
`testSyncOffsetForTargetGroupWithNullOffsetAndMetadata` cases, and a new
`testSyncOffsetForNewGroupOnTarget` test exercises the `DEAD` path that
creates
a group on the target by syncing the converted upstream offsets.
- [ ] Tests added or updated that fail before the change and pass after.
- [x] Tests added or updated for the new `GroupState`-based gating logic and
the snapshot reset in `refreshIdleConsumerGroupOffset` /
`syncGroupOffset`.
- [ ] Integration/system tests added or updated.
- [ ] Documentation updated.
- [ ] Compatibility considerations noted (no public API changes; a new
package-private constructor was added for testing).
JIRA: https://issues.apache.org/jira/browse/KAFKA-19794
--
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]
