fix(inkless:delete): do not sleep inside the topic purger tick [KC-349] by viktorsomogyi · Pull Request #792 · aiven/inkless

viktorsomogyi · 2026-09-08T14:14:28Z

KafkaScheduler uses scheduleAtFixedRate. An idle or error sleep inside run() queues missed ticks and the next work cycle bursts past max.batches.per.cycle / interval. Empty cycles return immediately. Errors set nextEligibleMs and later ticks skip until that time.

InklessDisklessTopicDeleteTest asserts the per-cycle cap from TopicPurger logs instead of a COUNT(*) poll.

KafkaScheduler uses scheduleAtFixedRate. An idle or error sleep inside run() queues missed ticks and the next work cycle bursts past max.batches.per.cycle / interval. Empty cycles return immediately. Errors set nextEligibleMs and later ticks skip until that time. InklessDisklessTopicDeleteTest asserts the per-cycle cap from TopicPurger logs instead of a COUNT(*) poll. Co-authored-by: Cursor <cursoragent@cursor.com>

giuseppelillo

LGTM. Merging this one, but we have the same issue on FileCleaner, could you also fix that one?

jeqo

The fix looks good. Added a few comments.

Also, a related finding is that FileCleaner may have a similar issue as it's sleeping and it's scheduled thread. We could handle it here or in a separate PR, as you prefer.

jeqo · 2026-09-09T09:58:08Z

-            LOGGER.error("Error while purging deleted logs, waiting for {}", Duration.ofMillis(backoff), e);
-            time.sleep(backoff);
+            nextEligibleMs = time.milliseconds() + backoff;
+            LOGGER.error("Error while purging deleted logs, retrying after {}", Duration.ofMillis(backoff), e);


nit:

Suggested change

LOGGER.error("Error while purging deleted logs, retrying after {}", Duration.ofMillis(backoff), e);

LOGGER.error("Error while purging deleted logs, skipping ticks for the next {}", Duration.ofMillis(backoff), e);

jeqo · 2026-09-09T10:05:17Z

            }

            attempts.set(0);
+            nextEligibleMs = 0L;


At this point nextEligibleMs is already in the past, so should be fine to delete this one

jeqo · 2026-09-09T10:07:47Z

+    // KafkaScheduler uses scheduleAtFixedRate. Sleeping inside run() queues missed ticks, and
+    // the next work cycle bursts past topic.purger.max.batches.per.cycle / interval. Skip the
+    // tick until this time instead.


nit: a shorter version

Suggested change

// KafkaScheduler uses scheduleAtFixedRate. Sleeping inside run() queues missed ticks, and

// the next work cycle bursts past topic.purger.max.batches.per.cycle / interval. Skip the

// tick until this time instead.

// Do not sleep in run(): KafkaScheduler uses scheduleAtFixedRate, and a sleep queues missed

// ticks that then fire back to back past topic.purger.max.batches.per.cycle.

jeqo · 2026-09-09T10:28:14Z

            assertEquals(0, beforeDelete.deletedLogCount(), "Live logs must not be stamped deleted");

-            admin.deleteTopics(Collections.singletonList(TOPIC_NAME)).all().get(30, TimeUnit.SECONDS);
+            try (LogCaptureAppender appender = LogCaptureAppender.createAndRegister(TopicPurger.class)) {


nit: pin the log level explicitly

Suggested change

try (LogCaptureAppender appender = LogCaptureAppender.createAndRegister(TopicPurger.class)) {

try (LogCaptureAppender appender = LogCaptureAppender.createAndRegister(TopicPurger.class)) {

appender.setClassLogger(TopicPurger.class, Level.INFO)

jeqo · 2026-09-09T10:48:56Z

+    /**
+     * One cycle cannot finish the drain: each broker deletes at most
+     * {@code topic.purger.max.batches.per.cycle} batches. Count saturated cycles from TopicPurger
+     * logs instead of a {@code COUNT(*)} poll, which can miss leftovers between 1s ticks.
+     */
+    private static void assertSaturatedAcrossMultipleCycles(LogCaptureAppender appender, long batchesAtDelete) {
+        long saturated = appender.getMessages().stream()
+            .filter(message -> message.contains("per-cycle cap reached"))
+            .count();
+        assertTrue(saturated >= batchesAtDelete - NUM_BROKERS,
+            "TopicPurger must hit the per-cycle cap across multiple cycles (max "
+                + MAX_BATCHES_PER_CYCLE + " per cycle); saturated=" + saturated
+                + " batchesAtDelete=" + batchesAtDelete
+                + " messages=" + appender.getMessages());


Maybe worth adding a doc for batchesAtDelete - NUM_BROKERS as it's not obvious where this is coming from.

We could also consider to move the saturated calculation to a wait if condition, to expect all brokers to do the logging:

final long expected = batchesAtDelete - 1; // every deleting cycle but the last hits the cap TestUtils.waitForCondition(() -> countSaturated(appender) >= expected, 10_000, () -> "TopicPurger must hit the per-cycle cap on every cycle but the last; saturated=" + countSaturated(appender) + " batchesAtDelete=" + batchesAtDelete + " messages=" + appender.getMessages()); private static long countSaturated(LogCaptureAppender appender) { return appender.getMessages().stream() .filter(message -> message.contains(TopicPurger.CAP_REACHED_LOG_FRAGMENT)) .count(); }

…9] (#792) KafkaScheduler uses scheduleAtFixedRate. An idle or error sleep inside run() queues missed ticks and the next work cycle bursts past max.batches.per.cycle / interval. Empty cycles return immediately. Errors set nextEligibleMs and later ticks skip until that time. InklessDisklessTopicDeleteTest asserts the per-cycle cap from TopicPurger logs instead of a COUNT(*) poll. Co-authored-by: Cursor <cursoragent@cursor.com>

* docs(inkless:sync): archive the 0.48 cherry-pick sessions for 4.1, 4.2, 4.3 Session logs for bringing the three active release branches up to main (5fa6669, #784) ahead of the 0.48 release. inkless-4.3 took its 8 commits clean. inkless-4.2 and inkless-4.1 took 19 each; the logs record the conflict resolutions and the sync(compile)/sync(test) adaptations needed for the 4.2.1 and 4.1.2 APIs (Record package, LogReadResult constructor, Seq-based DelayedFetch statuses). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> docs(inkless:sync): complete the 0.48 cherry-pick session logs The three branches took six more main commits (#794, #787, #792, #793, tables and the verification that covered them. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * docs(inkless:release): list inkless-4.3 as an active release branch inkless-4.3 is on Kafka 4.3.1 (#779) and gets its first Kafka-base tag with 0.48, so it moves from Planned to Active in RELEASES.md. The release workflow header names it among the branches the cherry-pick sync has to cover before a run. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * docs(inkless:release): add 0.48 changelog entry Generated from inkless-release-0.47..inkless-release-0.48 and curated: every metric delta traced to its commit, the four config default changes the configs.rst diff shows added next to the four new keys, and the V28 lock on logs described for operators. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

viktorsomogyi force-pushed the svv/topic-purger-flaky branch from 9cdb257 to a7df04b Compare September 8, 2026 14:18

viktorsomogyi marked this pull request as ready for review September 9, 2026 07:58

viktorsomogyi force-pushed the svv/topic-purger-flaky branch from a7df04b to 824f2f3 Compare September 9, 2026 08:31

viktorsomogyi requested a review from jeqo September 9, 2026 08:52

giuseppelillo approved these changes Sep 9, 2026

View reviewed changes

giuseppelillo merged commit 5693b9d into main Sep 9, 2026
5 checks passed

giuseppelillo deleted the svv/topic-purger-flaky branch September 9, 2026 09:49

8000

jeqo reviewed Sep 9, 2026

View reviewed changes

viktorsomogyi mentioned this pull request Sep 9, 2026

fix(inkless:delete): do not sleep inside the file cleaner tick [KC-349] #795

Open

Navigation Menu

Repository navigation

fix(inkless:delete): do not sleep inside the topic purger tick [KC-349] - #792

fix(inkless:delete): do not sleep inside the topic purger tick [KC-349]#792
giuseppelillo merged 1 commit into
mainfrom
svv/topic-purger-flaky

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

	LOGGER.error("Error while purging deleted logs, retrying after {}", Duration.ofMillis(backoff), e);
	LOGGER.error("Error while purging deleted logs, skipping ticks for the next {}", Duration.ofMillis(backoff), e);

	try (LogCaptureAppender appender = LogCaptureAppender.createAndRegister(TopicPurger.class)) {
	try (LogCaptureAppender appender = LogCaptureAppender.createAndRegister(TopicPurger.class)) {
	appender.setClassLogger(TopicPurger.class, Level.INFO)

Conversation

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants