frankvicky commented on code in PR #22622:
URL: https://github.com/apache/kafka/pull/22622#discussion_r3460904936
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java:
##########
@@ -68,29 +80,278 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
private final Optional<StreamsGroupTopologyDescriptionPlugin> plugin;
private final StreamsGroupTopologyDescriptionBackoff backoff;
+ private final Timer timer;
+ private final long cleanupCheckIntervalMs;
+ private final Function<String, TopicPartition> topicPartitionFor;
+ private final GroupCoordinatorMetrics groupCoordinatorMetrics;
+
+ /**
+ * True between {@link #startCleanupCycle} and {@link #close}. Read at
every cycle
+ * boundary that would otherwise schedule new plugin calls or runtime
writes, so a
+ * cycle that is in flight when {@code close} fires drains rather than
racing the
+ * runtime tear-down.
+ */
+ private final AtomicBoolean running = new AtomicBoolean(false);
+
+ /**
+ * Single-flight guard for the periodic cleanup cycle: a tick that fires
while the
+ * previous cycle is still settling per-group plugin calls and
conditional-clear writes
+ * is dropped. Set true at the top of {@link #runCleanupCycle}, released
by the
+ * terminal {@code whenComplete} that joins all per-partition reads and
per-group
+ * futures.
+ */
+ private final AtomicBoolean cycleInFlight = new AtomicBoolean(false);
+
+ /**
+ * The currently-scheduled cleanup tick on the broker-level {@link Timer}.
+ * Self-rescheduled inside the {@link TimerTask}'s {@code run}; {@link
#close} cancels
+ * this snapshot and the task's own re-arm check observes {@code running
== false}
+ * so the next tick does not re-schedule itself.
+ */
+ private volatile TimerTask scheduledTask;
+
public StreamsGroupTopologyDescriptionManager(
LogContext logContext,
Optional<StreamsGroupTopologyDescriptionPlugin> plugin,
- Time time
+ Time time,
+ Timer timer,
+ long cleanupCheckIntervalMs,
Review Comment:
One conflict worth flagging: your earlier feedback
(`recordPluginDeleteOutcome` is a duplicate, should go into the manager) and
this one (`GroupCoordinatorMetrics` stays on the service) can't both hold here.
With `GroupCoordinatorMetrics` out of the manager,
`invokeDeleteTopologies`can't record on its own, so the two callers (cycle
body, `deleteStreamsTopologyDescriptions`) both call
`service.recordPluginDeleteOutcome(...)` again. I went with the split you
described in this comment as the priority since it's the more recent guidance,
but happy to fold recording back into the manager (constructor would grow by
`GroupCoordinatorMetrics`) if you'd rather keep the de-dup.
--
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]
