lucasbru commented on code in PR #22622:
URL: https://github.com/apache/kafka/pull/22622#discussion_r3466095679
##########
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:
You've read the conflict correctly. Keeping GroupCoordinatorMetrics out of
the manager is the priority — the duplication of the private helper is fine.
The split you went with is the right call.
--
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]
