diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 4ca9573d74..2512c78ad8 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -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} @@ -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 @@ -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" @@ -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] = @@ -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) @@ -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) + } + } + + // 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() + } + } 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 @@ -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") diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala index 27b7734f32..a2a8b5c6a8 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala @@ -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 @@ -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 = { @@ -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) { @@ -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()) @@ -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) diff --git a/storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java b/storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java index 78bf92a5a6..caabf1ce86 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java @@ -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; @@ -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; @@ -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 noWorkBackoffSupplier; - /** - * The counter of cleaning attempts. - */ - private final AtomicInteger attempts = new AtomicInteger(); + private final AtomicBoolean closed = new AtomicBoolean(false); public FileCleaner(SharedState sharedState) { this( @@ -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); @@ -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 { if (saturated) { metrics.recordFileCleanerCycleSaturated(); @@ -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); } } @@ -170,7 +158,9 @@ private int cleanFiles(Set 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(); + } } } diff --git a/storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcer.java b/storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcer.java index 61028ce1df..4a08473248 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcer.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcer.java @@ -18,6 +18,7 @@ package io.aiven.inkless.delete; import org.apache.kafka.common.TopicIdPartition; +import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.utils.Time; import org.apache.kafka.storage.internals.log.LogConfig; @@ -34,6 +35,7 @@ import java.util.Map; import java.util.Objects; import java.util.Random; +import java.util.concurrent.atomic.AtomicBoolean; import io.aiven.inkless.TimeUtils; import io.aiven.inkless.common.SharedState; @@ -50,8 +52,8 @@ public class RetentionEnforcer implements Runnable, Closeable { private final ControlPlane controlPlane; private final RetentionEnforcementScheduler retentionEnforcementScheduler; private final int maxBatchesPerRequest; - private final RetentionEnforcerMetrics metrics; + private final AtomicBoolean closed = new AtomicBoolean(false); public RetentionEnforcer(final SharedState sharedState) { this(Objects.requireNonNull(sharedState, "sharedState cannot be null").time(), @@ -83,6 +85,9 @@ public RetentionEnforcer(final SharedState sharedState) { @Override public void run() { + if (closed.get()) { + return; + } try { runUnsafe(); } catch (final Exception e) { @@ -121,54 +126,69 @@ private synchronized void runUnsafe() { LOGGER.info("Enforcing retention for {} partitions", requests.size()); final Instant start = TimeUtils.durationMeasurementNow(time); - final List responses; - try { - responses = controlPlane.enforceRetention(requests, maxBatchesPerRequest); - } catch (final Exception e) { - metrics.recordRetentionEnforcementFinishedWithError(); - LOGGER.error("Unexpected error when enforcing retention", e); - throw new RuntimeException(e); - } - final Instant ended = TimeUtils.durationMeasurementNow(time); - final long durationMs = Duration.between(start, ended).toMillis(); - long totalBatchesDeleted = 0; long totalBytesDeleted = 0; - for (int i = 0; i < responses.size(); i++) { - final TopicIdPartition enforcedPartition = enforcedPartitions.get(i); - final var request = requests.get(i); - final var response = responses.get(i); - - switch (response.errors()) { - case NONE -> { - LOGGER.trace("Enforcing retention for {} with retentionBytes={}, retentionMs={} completed successfully. " + - "{} batches and {} bytes deleted, new log start offset: {}", - enforcedPartition.topicPartition(), request.retentionBytes(), request.retentionMs(), - response.batchesDeleted(), response.bytesDeleted(), response.logStartOffset()); + int partitionsEnforced = 0; + // One call per partition creates a shutdown boundary; the control plane already uses one + // transaction per partition. + for (int i = 0; i < requests.size(); i++) { + if (closed.get()) { + LOGGER.info("Retention enforcement stopping after {} of {} partitions: enforcer closed", + partitionsEnforced, requests.size()); + break; + } + final EnforceRetentionRequest request = requests.get(i); + final List responses; + try { + responses = controlPlane.enforceRetention(List.of(request), maxBatchesPerRequest); + } catch (final Exception e) { + metrics.recordRetentionEnforcementFinishedWithError(); + throw e; + } + if (!responses.isEmpty()) { + final EnforceRetentionResponse response = responses.get(0); + logResponse(request, enforcedPartitions.get(i), response); + if (response.errors() == Errors.NONE) { totalBatchesDeleted += response.batchesDeleted(); totalBytesDeleted += response.bytesDeleted(); } - - case UNKNOWN_TOPIC_OR_PARTITION -> - // When a topic is deleted, each partition may still be attempted once. - // Use debug logging to not pollute the log with this normal behavior. - LOGGER.debug("Enforcing retention for {} with retentionBytes={}, retentionMs={} completed with error: {}", - enforcedPartition.topicPartition(), request.retentionBytes(), request.retentionMs(), response.errors()); - - default -> - LOGGER.error("Enforcing retention for {} with retentionBytes={}, retentionMs={} completed with error: {}", - enforcedPartition.topicPartition(), request.retentionBytes(), request.retentionMs(), response.errors()); } + partitionsEnforced = i + 1; } + final Instant ended = TimeUtils.durationMeasurementNow(time); + final long durationMs = Duration.between(start, ended).toMillis(); metrics.recordRetentionEnforcementFinishedSuccessfully(durationMs, totalBatchesDeleted, totalBytesDeleted); LOGGER.info("Enforced retention for {} partitions in {} ms: {} batches, {} bytes deleted", - requests.size(), durationMs, totalBatchesDeleted, totalBytesDeleted); + partitionsEnforced, durationMs, totalBatchesDeleted, totalBytesDeleted); + } + + private void logResponse(final EnforceRetentionRequest request, + final TopicIdPartition enforcedPartition, + final EnforceRetentionResponse response) { + switch (response.errors()) { + case NONE -> + LOGGER.trace("Enforcing retention for {} with retentionBytes={}, retentionMs={} completed successfully. " + + "{} batches and {} bytes deleted, new log start offset: {}", + enforcedPartition.topicPartition(), request.retentionBytes(), request.retentionMs(), + response.batchesDeleted(), response.bytesDeleted(), response.logStartOffset()); + + case UNKNOWN_TOPIC_OR_PARTITION -> + // A deleted topic's partitions may still be attempted once. + LOGGER.debug("Enforcing retention for {} with retentionBytes={}, retentionMs={} completed with error: {}", + enforcedPartition.topicPartition(), request.retentionBytes(), request.retentionMs(), response.errors()); + + default -> + LOGGER.error("Enforcing retention for {} with retentionBytes={}, retentionMs={} completed with error: {}", + enforcedPartition.topicPartition(), request.retentionBytes(), request.retentionMs(), response.errors()); + } } @Override public void close() throws IOException { - metrics.close(); + if (closed.compareAndSet(false, true)) { + metrics.close(); + } } } diff --git a/storage/inkless/src/test/java/io/aiven/inkless/delete/FileCleanerMockedTest.java b/storage/inkless/src/test/java/io/aiven/inkless/delete/FileCleanerMockedTest.java index 77d6fdc952..07ee072daf 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/delete/FileCleanerMockedTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/delete/FileCleanerMockedTest.java @@ -43,13 +43,16 @@ import io.aiven.inkless.storage_backend.common.StorageBackend; import io.aiven.inkless.storage_backend.common.StorageBackendException; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anySet; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) @@ -255,4 +258,51 @@ void doesNotRecordSaturatedCycleBelowMaxFilesPerCycle() throws Exception { assertEquals(0, cleaner.metrics.fileCleanerCycleSaturated.sum()); } + + @Test + void noWorkDoesNotSleepOnThread() { + final var cleaner = new FileCleaner(time, controlPlane, storageBackend, OBJECT_KEY_CREATOR, RETENTION_PERIOD, MAX_FILES_PER_CYCLE); + when(controlPlane.getFilesToDelete(any(), anyInt())).thenReturn(List.of()); + + final long before = time.milliseconds(); + cleaner.run(); + assertEquals(before, time.milliseconds()); + } + + @Test + void errorDoesNotSleepOnThreadOrPropagate() { + final var cleaner = new FileCleaner(time, controlPlane, storageBackend, OBJECT_KEY_CREATOR, RETENTION_PERIOD, MAX_FILES_PER_CYCLE); + when(controlPlane.getFilesToDelete(any(), anyInt())).thenThrow(new RuntimeException("boom")); + + final long before = time.milliseconds(); + assertDoesNotThrow(cleaner::run); + assertEquals(before, time.milliseconds()); + } + + @Test + void runAfterCloseIsNoOp() throws Exception { + final var cleaner = new FileCleaner(time, controlPlane, storageBackend, OBJECT_KEY_CREATOR, RETENTION_PERIOD, MAX_FILES_PER_CYCLE); + cleaner.close(); + + cleaner.run(); + + verifyNoInteractions(controlPlane, storageBackend); + } + + @Test + void closeDuringCycleSkipsDeletion() throws Exception { + final var cleaner = new FileCleaner(time, controlPlane, storageBackend, OBJECT_KEY_CREATOR, RETENTION_PERIOD, MAX_FILES_PER_CYCLE); + final var objectKey = OBJECT_KEY_CREATOR.from("key"); + final var now = TimeUtils.now(time); + when(controlPlane.getFilesToDelete(any(), anyInt())).thenAnswer(invocation -> { + cleaner.close(); + return List.of(new FileToDelete(objectKey.value(), now.minus(Duration.ofMinutes(15)))); + }); + + cleaner.run(); + + verify(storageBackend, times(0)).delete(anySet()); + verify(controlPlane, times(0)).deleteFiles(any()); + } + } diff --git a/storage/inkless/src/test/java/io/aiven/inkless/delete/RetentionEnforcerTest.java b/storage/inkless/src/test/java/io/aiven/inkless/delete/RetentionEnforcerTest.java index 6004b59672..5e74fd6835 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/delete/RetentionEnforcerTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/delete/RetentionEnforcerTest.java @@ -47,7 +47,9 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) @@ -149,10 +151,41 @@ void onlyDeleteTopics() throws Exception { try (final var enforcer = new RetentionEnforcer(time, metadataView, controlPlane, retentionEnforcementScheduler, 0)) { enforcer.run(); - verify(controlPlane).enforceRetention(requestCaptor.capture(), eq(0)); - assertThat(requestCaptor.getValue()) + verify(controlPlane, times(2)).enforceRetention(requestCaptor.capture(), eq(0)); + assertThat(requestCaptor.getAllValues()).allSatisfy(rs -> assertThat(rs).hasSize(1)); + assertThat(requestCaptor.getAllValues()) + .flatExtracting(rs -> rs) .map(EnforceRetentionRequest::topicId) .containsExactly(TOPIC_ID_1, TOPIC_ID_2); } } + + @Test + void closeStopsAtPartitionBoundary() throws Exception { + when(retentionEnforcementScheduler.getReadyPartitions()).thenReturn(List.of(T0P0, T1P0, T2P0)); + when(metadataView.getTopicConfig(any())).thenReturn(new LogConfig(Map.of())); + + final var enforcer = new RetentionEnforcer(time, metadataView, controlPlane, retentionEnforcementScheduler, 0); + when(controlPlane.enforceRetention(any(), eq(0))).thenAnswer(invocation -> { + enforcer.close(); + return List.of(); + }); + + enforcer.run(); + + verify(controlPlane, times(1)).enforceRetention(requestCaptor.capture(), eq(0)); + assertThat(requestCaptor.getValue()) + .map(EnforceRetentionRequest::topicId).containsExactly(TOPIC_ID_0); + } + + @Test + void runAfterCloseIsNoOp() throws Exception { + final var enforcer = new RetentionEnforcer(time, metadataView, controlPlane, retentionEnforcementScheduler, 0); + enforcer.close(); + + enforcer.run(); + + verifyNoInteractions(controlPlane, retentionEnforcementScheduler); + } + }