frankvicky opened a new pull request, #22622:
URL: https://github.com/apache/kafka/pull/22622
JIRA: KAFKA-20696
This PR is a part of KIP-1331
Cycle (broker-level, on GroupCoordinatorService):
- A self-rescheduling TimerTask runs at offsets.retention.check.interval.ms,
gated on isActive + manager.isPluginConfigured(). Single-flight guarded so
a
cycle that fires while the previous one is still settling per-group futures
is dropped with a warn log.
- Each tick calls runtime.scheduleReadAllOperation to fan out an eligibility
query to every hosted shard, then fans plugin.deleteTopology out via the
existing manager.invokeDeleteTopologies building block. The manager itself
is unchanged — it keeps the (plugin, time) shape KAFKA-20623-4 settled on
and exposes only plugin-invocation + back-off mutations; chain assembly
lives on the service.
Eligibility + gating (per-shard):
- New GroupCoordinatorShard.listStreamsGroupsNeedingTopologyCleanup returns
group ids that are STREAMS + isEmpty + storedDescriptionTopologyEpoch != -1
+ offsets all past retention, paired with the observed storedEpoch. The
scan uses a non-throwing GroupMetadataManager.maybeGroup lookup to avoid
per-non-streams-group exception cost on plugin-enabled brokers running
mixed groups; per-group errors are logged and the scan continues.
- cleanupGroupMetadata now defers maybeDeleteGroup for STREAMS groups with
storedEpoch != -1, but only when a plugin is configured on this broker —
otherwise no cycle would ever clear the field and the gate would block
natural expiration indefinitely. The cast is factored into a named helper
so the short-circuit precondition for (StreamsGroup) sits next to the cast.
- After a successful plugin.deleteTopology, the cycle writes a conditional
GroupMetadataManager.clearStoredDescriptionTopologyEpoch(groupId,
expectedStoredEpoch) — only clears when the persisted value still equals
the epoch we observed at scan time, so a concurrent setTopology that has
advanced storedEpoch is preserved. Failed plugin calls retry on the next
cycle.
- The empty-group → concurrent-setTopology → cycle-delete race is documented
on runStreamsGroupTopologyCleanupCycle: plugin.deleteTopology keys only on
groupId per KIP appendix, so a new member that pushes between scan and
delete loses its plugin row; describe surfaces NOT_STORED + WARN, which is
the graceful-degradation path KIP-1331 explicitly accepts as "plugin-side
data loss".
OffsetMetadataManager:
- Adds a read-only allOffsetsExpired(groupId, currentTimestampMs) used by the
shard scan. The per-offset expirability check is factored into a private
isOffsetExpirable helper shared with cleanupExpiredOffsets so the read and
write paths cannot drift.
Metrics (KIP-1331):
- streams-group-topology-description-cleanup-cycle-{rate,count}
- streams-group-topology-description-cleanup-eligible-{rate,count}
- streams-group-topology-description-delete-{success,error}-{rate,count}
--
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]
