Skip to content

refactor(inkless): stop retention and cleanup jobs during broker shutdown [KC-417] #739

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

Draft
jeqo wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

refactor(inkless): stop retention and cleanup jobs during broker shutdown [KC-417] #739

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
54 changes: 47 additions & 7 deletions core/src/main/scala/kafka/server/ReplicaManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
package kafka.server

import com.yammer.metrics.core.Meter
import io.aiven.inkless.common.SharedState
import io.aiven.inkless.common.{InklessThreadFactory, SharedState}
import io.aiven.inkless.consume.{ConcatenatedRecords, FetchHandler, FetchOffsetHandler, Reader}
import io.aiven.inkless.storage_backend.common.ObjectFetcher
import io.aiven.inkless.control_plane.{AdvanceCrossTierLogStartOffsetRequest, AdvanceCrossTierLogStartOffsetResponse, BatchInfo, FindBatchRequest, FindBatchResponse, InitDisklessLogProducerState, RepairDisklessLogRequest, ListOffsetsRequest => CpListOffsetsRequest}
Expand Down Expand Up @@ -84,7 +84,7 @@ import java.nio.ByteBuffer
import java.nio.file.{Files, Paths}
import java.util
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.{CompletableFuture, ConcurrentHashMap, Future, RejectedExecutionException, TimeUnit}
import java.util.concurrent.{CompletableFuture, ConcurrentHashMap, Executors, Future, RejectedExecutionException, ScheduledExecutorService, TimeUnit}
import java.util.{Collections, Optional, OptionalInt, OptionalLong}
import java.util.function.Consumer
import java.util.stream.Collectors
Expand Down Expand Up @@ -147,6 +147,8 @@ object HostedPartition {
object ReplicaManager {
val HighWatermarkFilename = "replication-offset-checkpoint"

private val InklessBackgroundJobShutdownGraceMs = 10000L

private val LeaderCountMetricName = "LeaderCount"
private val PartitionCountMetricName = "PartitionCount"
private val OfflineReplicaCountMetricName = "OfflineReplicaCount"
Expand Down Expand Up @@ -310,6 +312,8 @@ class ReplicaManager(val config: KafkaConfig,
private val inklessDeleteRecordsInterceptor: Option[DeleteRecordsInterceptor] = inklessSharedState.map(new DeleteRecordsInterceptor(_))
private val inklessRetentionEnforcer: Option[RetentionEnforcer] = inklessSharedState.map(new RetentionEnforcer(_))
private val inklessFileCleaner: Option[FileCleaner] = inklessSharedState.map(new FileCleaner(_))
private val inklessBackgroundJobScheduler: Option[ScheduledExecutorService] =
inklessSharedState.map(_ => Executors.newScheduledThreadPool(2, new InklessThreadFactory("inkless-background-job-", true)))

// --- Diskless Partition Consolidation Fields ---
private val inklessConsolidatedDisklessLogPruner: Option[ConsolidatedDisklessLogPruner] =
Expand Down Expand Up @@ -509,9 +513,16 @@ class ReplicaManager(val config: KafkaConfig,

// Inkless threads
inklessSharedState.map { sharedState =>
scheduler.schedule("inkless-retention-enforcer", () => inklessRetentionEnforcer.foreach(_.run()), config.logInitialTaskDelayMs, 500L) // the real interval is inside

scheduler.schedule("inkless-file-cleaner", () => inklessFileCleaner.foreach(_.run()), sharedState.config().fileCleanerInterval().toMillis, sharedState.config().fileCleanerInterval().toMillis)
inklessBackgroundJobScheduler.foreach { bgScheduler =>
bgScheduler.scheduleAtFixedRate(
() => runInklessBackgroundJob("inkless-retention-enforcer", () => inklessRetentionEnforcer.foreach(_.run())),
config.logInitialTaskDelayMs, 500L, TimeUnit.MILLISECONDS) // the real interval is inside

val fileCleanerIntervalMs = sharedState.config().fileCleanerInterval().toMillis
bgScheduler.scheduleAtFixedRate(
() => runInklessBackgroundJob("inkless-file-cleaner", () => inklessFileCleaner.foreach(_.run())),
fileCleanerIntervalMs, fileCleanerIntervalMs, TimeUnit.MILLISECONDS)
}

// The default 30s task delay would leave EARLIEST wrong for up to 30s after every startup.
scheduler.schedule("inkless-cross-tier-log-start-reporter", () => sharedState.crossTierLogStartReporter().run(), sharedState.config().crossTierLogStartReportInterval().toMillis, sharedState.config().crossTierLogStartReportInterval().toMillis)
Expand All @@ -523,6 +534,36 @@ class ReplicaManager(val config: KafkaConfig,
}
}

// Scheduled executors cancel a periodic task when its body throws.
private def runInklessBackgroundJob(name: String, job: () => Unit): Unit = {
try job()
catch {
// An interrupt here is the shutdown backstop doing its job, not a fault.
case _: InterruptedException => Thread.currentThread().interrupt()
case t: Throwable => error(s"Uncaught exception in inkless background job '$name'", t)
Comment thread
jeqo marked this conversation as resolved.
}
}

// Shared state stays open during the grace period. On timeout, bounded shutdown takes priority over
// completing an in-flight job.
private def shutdownInklessBackgroundJobs(): Unit = {
inklessBackgroundJobScheduler.foreach(_.shutdown())
inklessRetentionEnforcer.foreach(_.close())
inklessFileCleaner.foreach(_.close())
inklessBackgroundJobScheduler.foreach { bgScheduler =>
try {
if (!bgScheduler.awaitTermination(ReplicaManager.InklessBackgroundJobShutdownGraceMs, TimeUnit.MILLISECONDS)) {
warn("Inkless background jobs did not stop within the shutdown grace period; interrupting")
bgScheduler.shutdownNow()
Comment thread
jeqo marked this conversation as resolved.
}
} catch {
case _: InterruptedException =>
bgScheduler.shutdownNow()
Thread.currentThread().interrupt()
}
}
}

private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasNonOfflinePartition = allPartitions.values.asScala.exists {
case online: HostedPartition.Online => topic == online.partition.topic
Expand Down Expand Up @@ -3385,8 +3426,7 @@ class ReplicaManager(val config: KafkaConfig,
inklessAppendHandler.foreach(_.close())
inklessFetchHandler.foreach(_.close())
inklessFetchOffsetHandler.foreach(_.close())
inklessRetentionEnforcer.foreach(_.close())
inklessFileCleaner.foreach(_.close())
shutdownInklessBackgroundJobs()
inklessDeleteRecordsInterceptor.foreach(_.close())
inklessSharedState.foreach(_.close())
info("Shut down completely")
Expand Down
66 changes: 64 additions & 2 deletions core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import io.aiven.inkless.common.SharedState
import io.aiven.inkless.config.InklessConfig
import io.aiven.inkless.consolidation.{ConsolidatedDisklessLogPruner, ConsolidationFetcherManager}
import io.aiven.inkless.consume.{ConcatenatedRecords, FetchHandler, FetchOffsetHandler}
import io.aiven.inkless.control_plane.{AdvanceCrossTierLogStartOffsetResponse, BatchInfo, BatchMetadata, ControlPlane, ControlPlaneException, FindBatchResponse, RepairDisklessLogRequest, RepairDisklessLogResponse, DeleteRecordsResponse => CpDeleteRecordsResponse, ListOffsetsRequest => CpListOffsetsRequest, ListOffsetsResponse => CpListOffsetsResponse}
import io.aiven.inkless.control_plane.{AdvanceCrossTierLogStartOffsetResponse, BatchInfo, BatchMetadata, ControlPlane, ControlPlaneException, FileToDelete, FindBatchResponse, RepairDisklessLogRequest, RepairDisklessLogResponse, DeleteRecordsResponse => CpDeleteRecordsResponse, ListOffsetsRequest => CpListOffsetsRequest, ListOffsetsResponse => CpListOffsetsResponse}
import io.aiven.inkless.produce.AppendHandler
import kafka.cluster.Partition
import kafka.server.QuotaFactory.QuotaManagers
Expand Down Expand Up @@ -7662,6 +7662,62 @@ class ReplicaManagerInklessTest {
}
}

@Test
def testShutdownWaitsForInFlightBackgroundJobBeforeClosingSharedState(): Unit = {
// Regression guard for the ordering inside ReplicaManager#shutdown: shutdownInklessBackgroundJobs()
// must run to completion (join the background executor) before inklessSharedState.foreach(_.close())
// runs. A reorder that closes the shared state first would let a job still in flight observe closed
// storage/control-plane resources. Component tests on FileCleaner/RetentionEnforcer only cover their
// own cooperative stop flag; they don't exercise ReplicaManager's shutdown sequencing at all, so they
// pass unchanged even if this order is swapped back.
val controlPlane = mock(classOf[ControlPlane])
val jobEntered = new CountDownLatch(1)
val releaseJob = new CountDownLatch(1)
when(controlPlane.getFilesToDelete(any(), anyInt())).thenAnswer { _ =>
jobEntered.countDown()
assertTrue(releaseJob.await(10, TimeUnit.SECONDS), "test never released the blocked background job")
util.List.of[FileToDelete]()
}

var sharedState: SharedState = null
val replicaManager = createReplicaManager(
List(disklessTopicPartition.topic()),
controlPlane = Some(controlPlane),
// Fire the file cleaner almost immediately instead of waiting out the 5-minute default.
extraInklessConfig = Map(InklessConfig.FILE_CLEANER_INTERVAL_MS_CONFIG -> Integer.valueOf(1)),
sharedStateHook = ss => sharedState = ss,
)
replicaManager.startup()

assertTrue(jobEntered.await(10, TimeUnit.SECONDS), "background file cleaner job never started")

val shutdownThread = new Thread(() => replicaManager.shutdown(checkpointHW = false), "shutdown-test-thread")
shutdownThread.start()
try {
// Poll for the shutdown thread's stack to contain both frames, not just TIMED_WAITING: several
// earlier shutdown steps (purgatory shutdown, other timed waits) can park the thread too, and a
// bare thread-state check would let the test release the job before shutdown actually reaches
// this join, defeating the regression guard.
waitUntilTrue(() => isBlockedInInklessBackgroundJobsAwaitTermination(shutdownThread),
"shutdown thread never parked inside shutdownInklessBackgroundJobs' awaitTermination")

verify(controlPlane, times(1)).getFilesToDelete(any(), anyInt())
verify(sharedState, never()).close()
} finally {
releaseJob.countDown()
shutdownThread.join(TimeUnit.SECONDS.toMillis(10))
}

assertFalse(shutdownThread.isAlive, "shutdown did not complete after the background job was released")
verify(sharedState, times(1)).close()
}

private def isBlockedInInklessBackgroundJobsAwaitTermination(thread: Thread): Boolean = {
val stack = thread.getStackTrace
stack.exists(_.getMethodName == "awaitTermination") &&
stack.exists(e => e.getClassName == classOf[ReplicaManager].getName && e.getMethodName == "shutdownInklessBackgroundJobs")
}

private def setupHybridLeaderPartition(replicaManager: ReplicaManager,
topicIdPartition: TopicIdPartition,
localEndOffset: Long): Partition = {
Expand Down Expand Up @@ -7839,7 +7895,11 @@ class ReplicaManagerInklessTest {
initDisklessLogManager: Option[InitDisklessLogManager] = None,
delayedFetchPurgatory: Option[DelayedOperationPurgatory[DelayedFetch]] = None,
defaultLogConfig: Option[LogConfig] = None,
crossTierLogStartCache: Option[CrossTierLogStartCache] = None
crossTierLogStartCache: Option[CrossTierLogStartCache] = None,
extraInklessConfig: Map[String, AnyRef] = Map.empty,
// Runs after all internal SharedState stubbing, so a test can capture the mock instance (for
// example, to verify close() ordering) without duplicating the stubbing above.
sharedStateHook: SharedState => Unit = _ => ()
): ReplicaManager = {
val props = TestUtils.createBrokerConfig(1, logDirCount = 2)
if (disklessManagedReplicasEnabled || disklessRemoteStorageConsolidationEnabled) {
Expand All @@ -7866,6 +7926,7 @@ class ReplicaManagerInklessTest {
val inklessConfigMap = new util.HashMap[String, Object]()
// Disable lagging consumer feature — not relevant for these tests
inklessConfigMap.put("fetch.lagging.consumer.thread.pool.size", Integer.valueOf(0))
extraInklessConfig.foreach { case (k, v) => inklessConfigMap.put(k, v) }
when(sharedState.config()).thenReturn(new InklessConfig(inklessConfigMap))
when(sharedState.controlPlane()).thenReturn(controlPlane.getOrElse(mock(classOf[ControlPlane])))
when(sharedState.maybeLaggingFetchStorage()).thenReturn(Optional.empty())
Expand All @@ -7889,6 +7950,7 @@ class ReplicaManagerInklessTest {
when(inklessMetadata.isRemoteStorageEnabled(t)).thenReturn(true)
}
when(sharedState.metadata()).thenReturn(inklessMetadata)
sharedStateHook(sharedState)

val logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size)

Expand Down
38 changes: 14 additions & 24 deletions storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@
*/
package io.aiven.inkless.delete;

import org.apache.kafka.common.utils.ExponentialBackoff;
import org.apache.kafka.common.utils.Time;

import org.slf4j.Logger;
Expand All @@ -29,8 +28,7 @@
import java.time.Instant;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import io.aiven.inkless.TimeUtils;
Expand All @@ -53,13 +51,8 @@ public class FileCleaner implements Runnable, Closeable {
final Duration retentionPeriod;
final int maxFilesPerCycle;
final FileCleanerMetrics metrics;
private final ExponentialBackoff errorBackoff = new ExponentialBackoff(100, 2, 60 * 1000, 0.2);
private final Supplier<Long> noWorkBackoffSupplier;

/**
* The counter of cleaning attempts.
*/
private final AtomicInteger attempts = new AtomicInteger();
private final AtomicBoolean closed = new AtomicBoolean(false);

public FileCleaner(SharedState sharedState) {
this(
Expand All @@ -86,16 +79,14 @@ public FileCleaner(SharedState sharedState) {
this.retentionPeriod = retentionPeriod;
this.maxFilesPerCycle = maxFilesPerCycle;
this.metrics = new FileCleanerMetrics(time);

// This backoff is needed only for jitter, there's no exponent in it.
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 (closed.get()) {
return;
}
try {
final var now = TimeUtils.now(time);

Expand All @@ -113,10 +104,10 @@ public void run() {
.map(FileToDelete::objectKey)
.collect(Collectors.toSet());
if (objectKeyPaths.isEmpty()) {
final long sleepMillis = noWorkBackoffSupplier.get();
final Duration sleepDuration = Duration.ofMillis(sleepMillis);
LOGGER.info("No files to delete, sleeping for {}", sleepDuration);
time.sleep(sleepMillis);
LOGGER.debug("No files to delete");
} else if (closed.get()) {
// Leave marked files for the next cycle instead of starting deletion during shutdown.
LOGGER.info("Skipping deletion of {} files: file cleaner closed", objectKeyPaths.size());
} else {
Comment thread
jeqo marked this conversation as resolved.
if (saturated) {
metrics.recordFileCleanerCycleSaturated();
Expand All @@ -132,13 +123,10 @@ public void run() {
LOGGER.info("File cleaner deleted {} of {} files", deletedCount, objectKeyPaths.size());
}

attempts.set(0);
metrics.recordFileCleanerCycleSucceeded();
} catch (final Exception e) {
metrics.recordFileCleanerError();
final long backoff = errorBackoff.backoff(attempts.incrementAndGet());
LOGGER.error("Error while deleting files, waiting for {}", Duration.ofMillis(backoff), e);
time.sleep(backoff);
LOGGER.error("Error while deleting files", e);
}
}

Expand Down Expand Up @@ -170,7 +158,9 @@ private int cleanFiles(Set<String> objectKeyPaths) throws StorageBackendExceptio

@Override
public void close() throws IOException {
// SharedState owns the storage backend lifecycle; only close component metrics here.
metrics.close();
if (closed.compareAndSet(false, true)) {
// SharedState owns the storage backend lifecycle; only close component metrics here.
metrics.close();
}
}
}
Loading
Loading