8000 fix(inkless:delete): do not sleep inside the topic purger tick [KC-349] by viktorsomogyi · Pull Request #792 · aiven/inkless · GitHub
[go: up one dir, main page]
Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
giuseppelillo merged 1 commit into from
Sep 9, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
81 changes: 38 additions & 43 deletions core/src/test/java/kafka/server/InklessDisklessTopicDeleteTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.test.KafkaClusterTestKit;
import org.apache.kafka.common.test.TestKitNodes;
import org.apache.kafka.common.utils.LogCaptureAppender;
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
import org.apache.kafka.server.config.ServerConfigs;
import org.apache.kafka.test.TestUtils;
Expand All @@ -50,10 +51,8 @@
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
Expand All @@ -64,6 +63,7 @@
import io.aiven.inkless.config.InklessConfig;
import io.aiven.inkless.control_plane.postgres.PostgresControlPlane;
import io.aiven.inkless.control_plane.postgres.PostgresControlPlaneConfig;
import io.aiven.inkless.delete.TopicPurger;
import io.aiven.inkless.storage_backend.s3.S3Storage;
import io.aiven.inkless.storage_backend.s3.S3StorageConfig;
import io.aiven.inkless.test_utils.InklessPostgreSQLContainer;
Expand Down Expand Up @@ -137,7 +137,7 @@ public void setup(final TestInfo testInfo) throws Exception {
.setConfigProp(InklessConfig.PREFIX + InklessConfig.PRODUCE_BUFFER_MAX_BYTES_CONFIG, Integer.toString(PRODUCE_BUFFER_MAX_BYTES))
// Drain and object delete must finish inside the wait windows below. The batch-coordinate
// cache TTL must stay at most half of file.cleaner.retention.period.ms. Cap the purger so
// the test can observe leftover batches between cycles.
// a single cycle cannot finish the drain.
.setConfigProp(InklessConfig.PREFIX + InklessConfig.TOPIC_PURGER_INTERVAL_MS_CONFIG, "1000")
.setConfigProp(InklessConfig.PREFIX + InklessConfig.TOPIC_PURGER_MAX_BATCHES_PER_CYCLE_CONFIG, Integer.toString(MAX_BATCHES_PER_CYCLE))
.setConfigProp(InklessConfig.PREFIX + InklessConfig.FILE_CLEANER_INTERVAL_MS_CONFIG, "1000")
Expand Down Expand Up @@ -185,20 +185,23 @@ public void testDeleteTopicsReturnsBeforeBackgroundPurge() throws Exception {
assertEquals(NUM_PARTITIONS, beforeDelete.logCount(), "Every partition must have a log row");
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)) {
@jeqo jeqo Sep 9, 2026
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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)

admin.deleteTopics(Collections.singletonList(TOPIC_NAME)).all().get(30, TimeUnit.SECONDS);

final ControlPlaneSnapshot afterDelete = readControlPlaneSnapshot(topicId);
log.info("Control plane snapshot immediately after delete: {}", afterDelete);
if (afterDelete.logCount() > 0) {
assertEquals(afterDelete.logCount(), afterDelete.deletedLogCount(),
"DELETE_TOPICS must stamp logs.deleted_at rather than drop the rows");
}
assertTrue(afterDelete.batchCount() >= MIN_BATCHES_FOR_MULTI_CYCLE,
"DELETE_TOPICS must return before TopicPurger drains the batches, got "
+ afterDelete.batchCount());
final ControlPlaneSnapshot afterDelete = readControlPlaneSnapshot(topicId);
log.info("Control plane snapshot immediately after delete: {}", afterDelete);
if (afterDelete.logCount() > 0) {
assertEquals(afterDelete.logCount(), afterDelete.deletedLogCount(),
"DELETE_TOPICS must stamp logs.deleted_at rather than drop the rows");
}
assertTrue(afterDelete.batchCount() >= MIN_BATCHES_FOR_MULTI_CYCLE,
"DELETE_TOPICS must return before TopicPurger drains the batches, got "
+ afterDelete.batchCount());

waitUntilPurged(topicId);
assertSaturatedAcrossMultipleCycles(appender, afterDelete.batchCount());
}
waitUntilTopicGone(admin);
waitUntilPurgedAcrossMultipleCycles(topicId, afterDelete.batchCount());
}

try (S3Client s3 = s3Container.getS3Client()) {
Expand Down Expand Up @@ -275,40 +278,32 @@ private void waitUntilTopicGone(Admin admin) throws InterruptedException {
}, 60_000, () -> "Topic must disappear from Kafka metadata after DELETE_TOPICS");
}

/**
* Polls until the control-plane rows are gone and requires a leftover batch count in between.
* With {@code topic.purger.max.batches.per.cycle} capped, one cycle cannot finish the drain.
*/
private void waitUntilPurgedAcrossMultipleCycles(Uuid topicId, long batchesAtDelete) throws InterruptedException {
final List<Long> observedBatchCounts = new ArrayList<>();
observedBatchCounts.add(batchesAtDelete);
private void waitUntilPurged(Uuid topicId) throws InterruptedException {
final AtomicReference<ControlPlaneSnapshot> last = new AtomicReference<>();
boolean sawPartialDrain = false;
final long deadline = System.currentTimeMillis() + 90_000;

while (System.currentTimeMillis() < deadline) {
TestUtils.waitForCondition(() -> {
ControlPlaneSnapshot snapshot = readControlPlaneSnapshot(topicId);
last.set(snapshot);
long batches = snapshot.batchCount();
if (observedBatchCounts.get(observedBatchCounts.size() - 1) != batches) {
observedBatchCounts.add(batches);
ControlPlaneSnapshot previous = last.getAndSet(snapshot);
if (previous == null || !previous.equals(snapshot)) {
log.info("TopicPurger progress: {}", snapshot);
}
if (batches > 0 && batches < batchesAtDelete) {
sawPartialDrain = true;
}
if (snapshot.isFullyPurged()) {
break;
}
Thread.sleep(100);
}
return snapshot.isFullyPurged();
}, 90_000, () -> "TopicPurger must drop logs, batches, and producer_state; last seen: " + last.get());
}

assertTrue(last.get() != null && last.get().isFullyPurged(),
"TopicPurger must drop logs, batches, and producer_state; last seen: " + last.get()
+ "; batch counts: " + observedBatchCounts);
assertTrue(sawPartialDrain,
"TopicPurger must drain batches across multiple cycles (max "
+ MAX_BATCHES_PER_CYCLE + " per cycle); observed batch counts: " + observedBatchCounts);
/**
* 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());
Comment on lines +293 to +306
@jeqo jeqo Sep 9, 2026
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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();
}

}

private record ControlPlaneSnapshot(
Expand Down
22 changes: 11 additions & 11 deletions storage/inkless/src/main/java/io/aiven/inkless/delete/TopicPurger.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@
import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

import io.aiven.inkless.TimeUtils;
import io.aiven.inkless.common.SharedState;
Expand All @@ -42,8 +41,11 @@ public class TopicPurger implements Runnable, Closeable {
final int maxBatchesPerCycle;
final TopicPurgerMetrics metrics;
private final ExponentialBackoff errorBackoff = new ExponentialBackoff(100, 2, 60 * 1000, 0.2);
private final Supplier<Long> noWorkBackoffSupplier;
private final AtomicInteger attempts = new AtomicInteger();
// 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.
Comment on lines +45 to +47
@jeqo jeqo Sep 9, 2026
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

private volatile long nextEligibleMs;

public TopicPurger(SharedState sharedState) {
this(
Expand All @@ -61,14 +63,13 @@ public TopicPurger(SharedState sharedState) {
this.controlPlane = controlPlane;
this.maxBatchesPerCycle = maxBatchesPerCycle;
this.metrics = new TopicPurgerMetrics(time);

final int noWorkBackoffDuration = 10 * 1000;
final var noWorkBackoff = new ExponentialBackoff(noWorkBackoffDuration, 1, noWorkBackoffDuration * 2, 0.2);
noWorkBackoffSupplier = () -> noWorkBackoff.backoff(1);
}

@Override
public void run() {
if (time.milliseconds() < nextEligibleMs) {
return;
}
try {
metrics.recordTopicPurgerStart();
final PurgeDeletedLogsResponse result = TimeUtils.measureDurationMs(time,
Expand All @@ -77,9 +78,7 @@ public void run() {

metrics.recordTopicPurgerWorkRemain(result.moreRemain());
if (result.isEmpty()) {
final long sleepMillis = noWorkBackoffSupplier.get();
LOGGER.info("No purge work this cycle, sleeping for {}", Duration.ofMillis(sleepMillis));
time.sleep(sleepMillis);
LOGGER.debug("No purge work this cycle");
} else {
final boolean saturated = result.capReached();
if (saturated) {
Expand All @@ -95,12 +94,13 @@ public void run() {
}

attempts.set(0);
nextEligibleMs = 0L;
@jeqo jeqo Sep 9, 2026
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

metrics.recordTopicPurgerCycleSucceeded();
} catch (final Exception e) {
metrics.recordTopicPurgerError();
final long backoff = errorBackoff.backoff(attempts.incrementAndGet());
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);
@jeqo jeqo Sep 9, 2026
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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);
-->
}
}

Expand Down
17 changes: 13 additions & 4 deletions storage/inkless/src/test/java/io/aiven/inkless/delete/TopicPurgerMockedTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@
import io.aiven.inkless.control_plane.PurgeDeletedLogsResponse;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
Expand All @@ -54,17 +53,19 @@ void empty() {

assertEquals(-1, purger.metrics.lastSuccessfulPurgeTimeMs.get());

final long beforeMs = time.milliseconds();
purger.run();

verify(controlPlane, times(1)).purgeDeletedLogs(eq(MAX_BATCHES_PER_CYCLE));
// A cycle with no work still counts: the gauge answers "is the purger running", not "is it deleting".
assertEquals(time.milliseconds(), purger.metrics.lastSuccessfulPurgeTimeMs.get());
assertEquals(0, purger.metrics.topicPurgerCycleSaturated.intValue());
assertEquals(0, purger.metrics.topicPurgerWorkRemain.get());
assertEquals(beforeMs, time.milliseconds());
}

@Test
void noWorkIsIdleEvenWhenMoreRemain() {
void noWorkDoesNotSleepWhenMoreRemain() {
final var purger = new TopicPurger(time, controlPlane, MAX_BATCHES_PER_CYCLE);
// Another broker holds the deleted rows (SKIP LOCKED miss): moreRemain is cluster-wide.
when(controlPlane.purgeDeletedLogs(MAX_BATCHES_PER_CYCLE))
Expand All @@ -73,7 +74,7 @@ void noWorkIsIdleEvenWhenMoreRemain() {
final long beforeMs = time.milliseconds();
purger.run();

assertTrue(time.milliseconds() > beforeMs);
assertEquals(beforeMs, time.milliseconds());
assertEquals(0, purger.metrics.topicPurgerCycleSaturated.intValue());
assertEquals(1, purger.metrics.topicPurgerWorkRemain.get());
assertEquals(time.milliseconds(), purger.metrics.lastSuccessfulPurgeTimeMs.get());
Expand Down Expand Up @@ -118,16 +119,24 @@ void unboundedDoesNotSaturate() {
}

@Test
void doesNotTrackFailedCycleAsSuccessful() {
void errorBackoffSkipsTicksUntilEligible() {
final var purger = new TopicPurger(time, controlPlane, MAX_BATCHES_PER_CYCLE);
when(controlPlane.purgeDeletedLogs(MAX_BATCHES_PER_CYCLE))
.thenThrow(new RuntimeException("purge failed"))
.thenReturn(PurgeDeletedLogsResponse.empty());

purger.run();
assertEquals(-1, purger.metrics.lastSuccessfulPurgeTimeMs.get());
verify(controlPlane, times(1)).purgeDeletedLogs(eq(MAX_BATCHES_PER_CYCLE));

purger.run();
assertEquals(-1, purger.metrics.lastSuccessfulPurgeTimeMs.get());
verify(controlPlane, times(1)).purgeDeletedLogs(eq(MAX_BATCHES_PER_CYCLE));

// First error backoff is 100 * 2^1 * [0.8, 1.2] ms. Jump past the max.
time.sleep(1_000);
purger.run();
assertEquals(time.milliseconds(), purger.metrics.lastSuccessfulPurgeTimeMs.get());
verify(controlPlane, times(2)).purgeDeletedLogs(eq(MAX_BATCHES_PER_CYCLE));
}
}
Loading
0