diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java index 1ac6028af516c..0c78e88db92f7 100644 --- a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java +++ b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java @@ -818,6 +818,7 @@ abstract class RLMTask extends CancellableRunnable { protected final TopicIdPartition topicIdPartition; private final Logger logger; + private final ReentrantLock executionLock = new ReentrantLock(); public RLMTask(TopicIdPartition topicIdPartition) { this.topicIdPartition = topicIdPartition; @@ -829,33 +830,48 @@ protected LogContext getLogContext() { } public void run() { - if (isCancelled()) { - logger.debug("Skipping the current run for partition {} as it is cancelled", topicIdPartition); - return; - } - if (!remoteLogMetadataManagerPlugin.get().isReady(topicIdPartition)) { - logger.debug("Skipping the current run for partition {} as the remote-log metadata is not ready", topicIdPartition); - return; - } - + executionLock.lock(); try { - Optional unifiedLogOptional = fetchLog.apply(topicIdPartition.topicPartition()); - - if (unifiedLogOptional.isEmpty()) { + if (isCancelled()) { + logger.debug("Skipping the current run for partition {} as it is cancelled", topicIdPartition); return; } - - execute(unifiedLogOptional.get()); - } catch (InterruptedException ex) { - if (!isCancelled()) { - logger.warn("Current thread for partition {} is interrupted", topicIdPartition, ex); + if (!remoteLogMetadataManagerPlugin.get().isReady(topicIdPartition)) { + logger.debug("Skipping the current run for partition {} as the remote-log metadata is not ready", topicIdPartition); + return; } - } catch (RetriableException | RetriableRemoteStorageException ex) { - logger.debug("Encountered a retryable error while executing current task for partition {}", topicIdPartition, ex); - } catch (Exception ex) { - if (!isCancelled()) { - logger.warn("Current task for partition {} received error but it will be scheduled", topicIdPartition, ex); + + try { + Optional unifiedLogOptional = fetchLog.apply(topicIdPartition.topicPartition()); + + if (unifiedLogOptional.isEmpty()) { + return; + } + + execute(unifiedLogOptional.get()); + } catch (InterruptedException ex) { + if (!isCancelled()) { + logger.warn("Current thread for partition {} is interrupted", topicIdPartition, ex); + } + } catch (RetriableException | RetriableRemoteStorageException ex) { + logger.debug("Encountered a retryable error while executing current task for partition {}", topicIdPartition, ex); + } catch (Exception ex) { + if (!isCancelled()) { + logger.warn("Current task for partition {} received error but it will be scheduled", topicIdPartition, ex); + } } + } finally { + executionLock.unlock(); + } + } + + void awaitExecutionCompletion() { + executionLock.lock(); + try { + // Wait for an in-flight execution to finish before its owner proceeds with cleanup. + logger.debug("Remote log task for partition {} completed", topicIdPartition); + } finally { + executionLock.unlock(); } } @@ -1528,8 +1544,8 @@ void cleanupExpiredRemoteLogSegments() throws RemoteStorageException, ExecutionE } } - // Update log start offset with the computed value after retention cleanup is done - remoteLogRetentionHandler.logStartOffset.ifPresent(offset -> handleLogStartOffsetUpdate(topicIdPartition.topicPartition(), offset)); + // Cancellation must win over log-start-offset updates. + if (!isCancelled()) remoteLogRetentionHandler.logStartOffset.ifPresent(offset -> handleLogStartOffsetUpdate(topicIdPartition.topicPartition(), offset)); // At this point in time we have updated the log start offsets, but not initiated a deletion. // Either a follower has picked up the changes to the log start offset, or they have not. @@ -2233,6 +2249,8 @@ public void cancel() { future.cancel(true); } catch (Exception ex) { LOGGER.error("Error occurred while canceling the task: {}", rlmTask, ex); + } finally { + rlmTask.awaitExecutionCompletion(); } } diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java index 9f1f15a887285..399411fa8bce9 100644 --- a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java +++ b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java @@ -110,9 +110,11 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.BiConsumer; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -2804,6 +2806,96 @@ public void testDeletionOnRetentionBreachedSegments(long retentionSize, } } + @Test + public void testExpirationDoesNotAdvanceLogStartOffsetAfterCancellation() + throws RemoteStorageException, ExecutionException, InterruptedException { + Map logProps = new HashMap<>(); + logProps.put("retention.bytes", 0L); + logProps.put("retention.ms", -1L); + when(mockLog.config()).thenReturn(new LogConfig(logProps)); + when(mockLog.topicPartition()).thenReturn(leaderTopicIdPartition.topicPartition()); + when(mockLog.logEndOffset()).thenReturn(100L); + + checkpoint.write(List.of(epochEntry0)); + LeaderEpochFileCache cache = new LeaderEpochFileCache(leaderTopicIdPartition.topicPartition(), checkpoint, scheduler); + when(mockLog.leaderEpochCache()).thenReturn(cache); + + List metadata = listRemoteLogSegmentMetadata( + leaderTopicIdPartition, 1, 100, 1024, List.of(epochEntry0), RemoteLogSegmentState.COPY_SEGMENT_FINISHED); + when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition)) + .thenAnswer(invocation -> metadata.iterator()); + + CountDownLatch scanPaused = new CountDownLatch(1); + CountDownLatch resumeScan = new CountDownLatch(1); + Iterator blockingIterator = new Iterator<>() { + private boolean hasReturnedMetadata = false; + + @Override + public boolean hasNext() { + if (hasReturnedMetadata) { + scanPaused.countDown(); + try { + assertTrue(resumeScan.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + return !hasReturnedMetadata; + } + + @Override + public RemoteLogSegmentMetadata next() { + hasReturnedMetadata = true; + return metadata.get(0); + } + }; + AtomicInteger epochSegmentListCalls = new AtomicInteger(); + when(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition, 0)) + .thenAnswer(invocation -> epochSegmentListCalls.getAndIncrement() == 0 + ? metadata.iterator() + : blockingIterator); + when(remoteLogMetadataManager.updateRemoteLogSegmentMetadata(any(RemoteLogSegmentMetadataUpdate.class))) + .thenReturn(CompletableFuture.completedFuture(null)); + + RemoteLogManager.RLMExpirationTask expirationTask = remoteLogManager.new RLMExpirationTask(leaderTopicIdPartition); + AtomicReference failure = new AtomicReference<>(); + Thread cleanupThread = new Thread(() -> { + try { + expirationTask.run(); + } catch (Throwable t) { + failure.set(t); + } + }); + cleanupThread.start(); + + Future scheduledFuture = mock(Future.class); + RemoteLogManager.RLMTaskWithFuture taskWithFuture = + new RemoteLogManager.RLMTaskWithFuture(expirationTask, scheduledFuture); + Thread cancellationThread = null; + try { + assertTrue(scanPaused.await(5, TimeUnit.SECONDS)); + cancellationThread = new Thread(taskWithFuture::cancel); + cancellationThread.start(); + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> { + while (!expirationTask.isCancelled()) { + Thread.yield(); + } + }); + assertTrue(cancellationThread.isAlive()); + } finally { + resumeScan.countDown(); + } + cancellationThread.join(5_000); + cleanupThread.join(5_000); + + assertFalse(cancellationThread.isAlive()); + assertFalse(cleanupThread.isAlive()); + assertNull(failure.get()); + verify(scheduledFuture).cancel(true); + assertEquals(0L, currentLogStartOffset.get()); + verify(remoteStorageManager, never()).deleteLogSegmentData(any()); + } + @Test public void testSizeRetentionDoesNotOverDeleteOnFreshLeaderUntilHighestRemoteOffsetSeeded() throws RemoteStorageException, ExecutionException, InterruptedException {