This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new ad0daa3962 [core] Quiesce manifest rewrite workers before cleanup
(#9376)
ad0daa3962 is described below
commit ad0daa39621a58e35f061cc95db43e3e8b688ca7
Author: QuakeWang <[email protected]>
AuthorDate: Wed Aug 26 10:32:26 2026 +0800
[core] Quiesce manifest rewrite workers before cleanup (#9376)
---
.../org/apache/paimon/utils/ThreadPoolUtils.java | 245 +++++++++++++++++++++
.../apache/paimon/utils/ThreadPoolUtilsTest.java | 238 ++++++++++++++++++++
.../paimon/operation/ManifestFileBlockMerger.java | 60 ++---
.../paimon/utils/ManifestReadThreadPool.java | 28 ++-
.../operation/ManifestRewriteCleanupTest.java | 181 +++++++++++++++
5 files changed, 718 insertions(+), 34 deletions(-)
diff --git
a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java
b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java
index 7ad3425367..b5c28a19a0 100644
--- a/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java
+++ b/paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java
@@ -24,11 +24,13 @@ import
org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -45,6 +47,13 @@ import static
org.apache.paimon.utils.ThreadUtils.newDaemonThreadFactory;
/** Utils for thread pool. */
public class ThreadPoolUtils {
+ /** An iterator which waits for its active batch to quiesce when closed. */
+ public interface CloseableBatchIterator<T> extends Iterator<T>,
AutoCloseable {
+
+ @Override
+ void close();
+ }
+
/**
* Create a thread pool with max thread number. Inactive threads will
automatically exit.
*
@@ -124,6 +133,22 @@ public class ThreadPoolUtils {
};
}
+ /**
+ * Parallel processes one bounded batch at a time and returns results in
input order.
+ *
+ * <p>The caller must close the iterator to cancel unstarted tasks and
wait for running tasks.
+ */
+ public static <T, U> CloseableBatchIterator<T>
sequentialBatchedExecuteCloseable(
+ ExecutorService executor,
+ Function<U, List<T>> processor,
+ List<U> input,
+ int queueSize) {
+ if (queueSize <= 0) {
+ throw new NegativeArraySizeException("queue size should not be
negative");
+ }
+ return new SequentialBatchIterator<>(executor, processor, input,
queueSize);
+ }
+
public static <U> void randomlyOnlyExecute(
ExecutorService executor, Consumer<U> processor, Collection<U>
input) {
awaitAllFutures(submitAllTasks(executor, processor, input));
@@ -194,4 +219,224 @@ public class ThreadPoolUtils {
}
}
}
+
+ private static class SequentialBatchIterator<T, U> implements
CloseableBatchIterator<T> {
+
+ private final ExecutorService executor;
+ private final Function<U, List<T>> processor;
+ private final Queue<List<U>> batches;
+ private final Queue<BatchTask<T, U>> activeTasks = new ArrayDeque<>();
+
+ private Iterator<T> activeResults =
Collections.<T>emptyList().iterator();
+ private T next;
+ private boolean closed;
+
+ private SequentialBatchIterator(
+ ExecutorService executor,
+ Function<U, List<T>> processor,
+ List<U> input,
+ int queueSize) {
+ this.executor = executor;
+ this.processor = processor;
+ this.batches = new ArrayDeque<>(Lists.partition(input, queueSize));
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (!closed) {
+ advanceIfNeeded();
+ }
+ return next != null;
+ }
+
+ @Override
+ public T next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ T result = next;
+ next = null;
+ return result;
+ }
+
+ private void advanceIfNeeded() {
+ while (next == null) {
+ if (activeResults.hasNext()) {
+ next = activeResults.next();
+ } else if (!activeTasks.isEmpty()) {
+ BatchTask<T, U> task = activeTasks.peek();
+ try {
+ List<T> results = task.result();
+ activeTasks.poll();
+ activeResults = results.iterator();
+ } catch (RuntimeException | Error failure) {
+ if (task.failureReported()) {
+ activeTasks.poll();
+ }
+ throw failure;
+ }
+ } else if (batches.isEmpty()) {
+ return;
+ } else {
+ submitBatch(batches.poll());
+ }
+ }
+ }
+
+ private void submitBatch(List<U> batch) {
+ ClassLoader classLoader =
Thread.currentThread().getContextClassLoader();
+ for (U input : batch) {
+ BatchTask<T, U> task = new BatchTask<>(processor, input,
classLoader);
+ executor.execute(task);
+ activeTasks.add(task);
+ }
+ }
+
+ @Override
+ public synchronized void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ batches.clear();
+
+ Throwable failure = null;
+ boolean interrupted = Thread.interrupted();
+ for (BatchTask<T, U> task : activeTasks) {
+ try {
+ task.cancel();
+ } catch (Throwable cleanupFailure) {
+ failure = firstOrSuppressed(cleanupFailure, failure);
+ }
+ }
+ for (BatchTask<T, U> task : activeTasks) {
+ while (true) {
+ try {
+ task.awaitCompletion();
+ break;
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ Throwable taskFailure = task.unreportedFailure();
+ if (taskFailure != null) {
+ failure = firstOrSuppressed(taskFailure, failure);
+ }
+ }
+ activeTasks.clear();
+ activeResults = Collections.<T>emptyList().iterator();
+ next = null;
+
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ if (failure != null) {
+ throw rethrow(failure);
+ }
+ }
+ }
+
+ private static class BatchTask<T, U> implements Runnable {
+
+ private static final int CREATED = 0;
+ private static final int RUNNING = 1;
+ private static final int CANCELLED = 2;
+ private static final int FINISHED = 3;
+
+ private final Function<U, List<T>> processor;
+ private final U input;
+ private final ClassLoader classLoader;
+ private final CountDownLatch completion = new CountDownLatch(1);
+
+ private int state = CREATED;
+ private Thread runner;
+ private List<T> result;
+ private Throwable failure;
+ private volatile boolean failureReported;
+
+ private BatchTask(Function<U, List<T>> processor, U input, ClassLoader
classLoader) {
+ this.processor = processor;
+ this.input = input;
+ this.classLoader = classLoader;
+ }
+
+ @Override
+ public void run() {
+ synchronized (this) {
+ if (state == CANCELLED) {
+ state = FINISHED;
+ completion.countDown();
+ return;
+ }
+ state = RUNNING;
+ runner = Thread.currentThread();
+ }
+
+ try {
+ Thread.currentThread().setContextClassLoader(classLoader);
+ result = processor.apply(input);
+ } catch (RuntimeException | Error taskFailure) {
+ failure = taskFailure;
+ } finally {
+ synchronized (this) {
+ runner = null;
+ state = FINISHED;
+ }
+ completion.countDown();
+ }
+ }
+
+ private synchronized void cancel() {
+ if (state == CREATED) {
+ state = CANCELLED;
+ completion.countDown();
+ } else if (state == RUNNING) {
+ runner.interrupt();
+ }
+ }
+
+ private List<T> result() {
+ try {
+ completion.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ if (failure != null) {
+ failureReported = true;
+ throw rethrow(failure);
+ }
+ return result;
+ }
+
+ private void awaitCompletion() throws InterruptedException {
+ completion.await();
+ }
+
+ private boolean failureReported() {
+ return failureReported;
+ }
+
+ private Throwable unreportedFailure() {
+ return failureReported ? null : failure;
+ }
+ }
+
+ private static Throwable firstOrSuppressed(Throwable newFailure, Throwable
previousFailure) {
+ if (previousFailure == null || previousFailure == newFailure) {
+ return newFailure;
+ }
+ previousFailure.addSuppressed(newFailure);
+ return previousFailure;
+ }
+
+ private static RuntimeException rethrow(Throwable failure) {
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ if (failure instanceof RuntimeException) {
+ return (RuntimeException) failure;
+ }
+ return new RuntimeException(failure);
+ }
}
diff --git
a/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java
b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java
new file mode 100644
index 0000000000..65a3331fc7
--- /dev/null
+++ b/paimon-api/src/test/java/org/apache/paimon/utils/ThreadPoolUtilsTest.java
@@ -0,0 +1,238 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.utils;
+
+import org.apache.paimon.utils.ThreadPoolUtils.CloseableBatchIterator;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowable;
+
+/** Tests for {@link ThreadPoolUtils}. */
+public class ThreadPoolUtilsTest {
+
+ @Test
+ public void testCloseableBatchReturnsInOrderAndBoundsSubmission() throws
Exception {
+ ThreadPoolExecutor workers = (ThreadPoolExecutor)
Executors.newFixedThreadPool(2);
+ ExecutorService consumer = Executors.newSingleThreadExecutor();
+ CountDownLatch firstStarted = new CountDownLatch(1);
+ CountDownLatch secondFinished = new CountDownLatch(1);
+ CountDownLatch releaseFirst = new CountDownLatch(1);
+ CloseableBatchIterator<Integer> iterator =
+ ThreadPoolUtils.sequentialBatchedExecuteCloseable(
+ workers,
+ input -> {
+ if (input == 0) {
+ firstStarted.countDown();
+ await(releaseFirst);
+ } else if (input == 1) {
+ secondFinished.countDown();
+ }
+ return Collections.singletonList(input);
+ },
+ Arrays.asList(0, 1, 2, 3),
+ 2);
+
+ try {
+ Future<Integer> firstResult =
+ consumer.submit(
+ () -> {
+ assertThat(iterator.hasNext()).isTrue();
+ return iterator.next();
+ });
+
+ assertThat(firstStarted.await(3, TimeUnit.SECONDS)).isTrue();
+ assertThat(secondFinished.await(3, TimeUnit.SECONDS)).isTrue();
+ assertThat(workers.getTaskCount()).isEqualTo(2);
+ assertThat(firstResult.isDone()).isFalse();
+
+ releaseFirst.countDown();
+ List<Integer> results = new ArrayList<>();
+ results.add(firstResult.get(3, TimeUnit.SECONDS));
+ assertThat(iterator.hasNext()).isTrue();
+ results.add(iterator.next());
+ assertThat(workers.getTaskCount()).isEqualTo(2);
+
+ assertThat(iterator.hasNext()).isTrue();
+ assertThat(workers.getTaskCount()).isEqualTo(4);
+ results.add(iterator.next());
+ assertThat(iterator.hasNext()).isTrue();
+ results.add(iterator.next());
+ assertThat(iterator.hasNext()).isFalse();
+ assertThat(results).containsExactly(0, 1, 2, 3);
+ } finally {
+ releaseFirst.countDown();
+ iterator.close();
+ consumer.shutdownNow();
+ workers.shutdownNow();
+ assertThat(consumer.awaitTermination(3,
TimeUnit.SECONDS)).isTrue();
+ assertThat(workers.awaitTermination(3, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
+ @Test
+ public void testCloseCancelsQueuedTasksAndWaitsUninterruptibly() throws
Exception {
+ ThreadPoolExecutor workers = (ThreadPoolExecutor)
Executors.newFixedThreadPool(1);
+ ExecutorService closer = Executors.newSingleThreadExecutor();
+ CountDownLatch secondStarted = new CountDownLatch(1);
+ CountDownLatch workerInterrupted = new CountDownLatch(1);
+ CountDownLatch allowSecondToExit = new CountDownLatch(1);
+ CountDownLatch closeStarted = new CountDownLatch(1);
+ AtomicInteger executions = new AtomicInteger();
+ AtomicBoolean thirdExecuted = new AtomicBoolean();
+ AtomicBoolean closeRestoredInterrupt = new AtomicBoolean();
+ AtomicReference<Thread> closeThread = new AtomicReference<>();
+ CloseableBatchIterator<Integer> iterator =
+ ThreadPoolUtils.sequentialBatchedExecuteCloseable(
+ workers,
+ input -> {
+ executions.incrementAndGet();
+ if (input == 1) {
+ secondStarted.countDown();
+ awaitIgnoringInterrupts(allowSecondToExit,
workerInterrupted);
+ } else if (input == 2) {
+ thirdExecuted.set(true);
+ }
+ return Collections.singletonList(input);
+ },
+ Arrays.asList(0, 1, 2),
+ 3);
+
+ try {
+ assertThat(iterator.hasNext()).isTrue();
+ assertThat(iterator.next()).isZero();
+ assertThat(secondStarted.await(3, TimeUnit.SECONDS)).isTrue();
+ assertThat(workers.getTaskCount()).isEqualTo(3);
+
+ Future<?> closeResult =
+ closer.submit(
+ () -> {
+ closeThread.set(Thread.currentThread());
+ closeStarted.countDown();
+ iterator.close();
+
closeRestoredInterrupt.set(Thread.currentThread().isInterrupted());
+ });
+ assertThat(closeStarted.await(3, TimeUnit.SECONDS)).isTrue();
+ assertThat(workerInterrupted.await(3, TimeUnit.SECONDS)).isTrue();
+ assertThat(closeResult.isDone()).isFalse();
+
+ closeThread.get().interrupt();
+ allowSecondToExit.countDown();
+ closeResult.get(3, TimeUnit.SECONDS);
+
+ assertThat(closeRestoredInterrupt).isTrue();
+ assertThat(executions).hasValue(2);
+ assertThat(thirdExecuted).isFalse();
+ assertThat(iterator.hasNext()).isFalse();
+ iterator.close();
+ iterator.close();
+ assertThat(executions).hasValue(2);
+ } finally {
+ allowSecondToExit.countDown();
+ iterator.close();
+ closer.shutdownNow();
+ workers.shutdownNow();
+ assertThat(closer.awaitTermination(3, TimeUnit.SECONDS)).isTrue();
+ assertThat(workers.awaitTermination(3, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
+ @Test
+ public void testClosePreservesPrimaryErrorAndSuppressesWorkerError()
throws Exception {
+ ThreadPoolExecutor workers = (ThreadPoolExecutor)
Executors.newFixedThreadPool(2);
+ CountDownLatch secondStarted = new CountDownLatch(1);
+ CountDownLatch waitForCancellation = new CountDownLatch(1);
+ AssertionError primaryFailure = new AssertionError("primary failure");
+ AssertionError workerFailure = new AssertionError("worker failure");
+ CloseableBatchIterator<Integer> iterator =
+ ThreadPoolUtils.sequentialBatchedExecuteCloseable(
+ workers,
+ input -> {
+ if (input == 0) {
+ await(secondStarted);
+ } else {
+ secondStarted.countDown();
+ try {
+ waitForCancellation.await();
+ } catch (InterruptedException e) {
+ throw workerFailure;
+ }
+ }
+ return Collections.singletonList(input);
+ },
+ Arrays.asList(0, 1),
+ 2);
+
+ try {
+ assertThat(iterator.hasNext()).isTrue();
+ assertThat(iterator.next()).isZero();
+
+ Throwable thrown =
+ catchThrowable(
+ () -> {
+ try (CloseableBatchIterator<Integer> ignored =
iterator) {
+ throw primaryFailure;
+ }
+ });
+
+
assertThat(thrown).isSameAs(primaryFailure).hasSuppressedException(workerFailure);
+ iterator.close();
+ } finally {
+ waitForCancellation.countDown();
+ iterator.close();
+ workers.shutdownNow();
+ assertThat(workers.awaitTermination(3, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static void awaitIgnoringInterrupts(CountDownLatch latch,
CountDownLatch interrupted) {
+ while (true) {
+ try {
+ latch.await();
+ return;
+ } catch (InterruptedException e) {
+ interrupted.countDown();
+ }
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java
index 05b6225f60..4f1bb922ab 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java
@@ -42,6 +42,7 @@ import org.apache.paimon.stats.SimpleStatsConverter;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.Filter;
+import org.apache.paimon.utils.ThreadPoolUtils.CloseableBatchIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,6 +61,7 @@ import java.util.function.Function;
import static org.apache.paimon.manifest.ManifestFileMeta.allContainsRowId;
import static
org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+import static
org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecuteCloseable;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkState;
@@ -398,33 +400,37 @@ final class ManifestFileBlockMerger {
e);
}
};
- for (ManifestRewritePlan plan :
- sequentialBatchedExecute(planner, manifests,
manifestReadParallelism)) {
- if (fullCompaction
- && mustChange != null
- && !mustChange.test(plan.manifest)
- && plan.unchanged()) {
- checkState(
- unchangedManifests != null,
- "Full compaction requires an unchanged
manifest result.");
- unchangedManifests.add(plan.manifest);
- continue;
- }
- for (PlannedBlock block : plan.blocks) {
- if (block.compaction.canCopyEncodedBlock()) {
- writer.writeEncodedBlock(
- block.raw.encodedBlock(),
block.compaction.metadata);
- } else {
- writeBlockEntries(
- block.raw,
- writer,
- deletes,
- reusableIdentifier,
- fullCompaction,
- matchedEntries,
- emittedDeletes,
- metadata,
- plan.encodedRecordsCompatible);
+ try (CloseableBatchIterator<ManifestRewritePlan> plans =
+ sequentialBatchedExecuteCloseable(
+ planner, manifests, manifestReadParallelism)) {
+ while (plans.hasNext()) {
+ ManifestRewritePlan plan = plans.next();
+ if (fullCompaction
+ && mustChange != null
+ && !mustChange.test(plan.manifest)
+ && plan.unchanged()) {
+ checkState(
+ unchangedManifests != null,
+ "Full compaction requires an unchanged
manifest result.");
+ unchangedManifests.add(plan.manifest);
+ continue;
+ }
+ for (PlannedBlock block : plan.blocks) {
+ if (block.compaction.canCopyEncodedBlock()) {
+ writer.writeEncodedBlock(
+ block.raw.encodedBlock(),
block.compaction.metadata);
+ } else {
+ writeBlockEntries(
+ block.raw,
+ writer,
+ deletes,
+ reusableIdentifier,
+ fullCompaction,
+ matchedEntries,
+ emittedDeletes,
+ metadata,
+ plan.encodedRecordsCompatible);
+ }
}
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java
b/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java
index 7cf9778714..0ef818762d 100644
---
a/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java
+++
b/paimon-core/src/main/java/org/apache/paimon/utils/ManifestReadThreadPool.java
@@ -57,19 +57,33 @@ public class ManifestReadThreadPool {
Function<U, List<T>> processor, List<U> input, @Nullable Integer
threadNum) {
threadNum = normalizeThreadNum(threadNum);
ExecutorService executor = getExecutorService(threadNum);
- if (threadNum == null) {
- threadNum =
- executor instanceof ThreadPoolExecutor
- ? ((ThreadPoolExecutor)
executor).getMaximumPoolSize()
- : ((SemaphoredDelegatingExecutor)
executor).getPermitCount();
- }
- return ThreadPoolUtils.sequentialBatchedExecute(executor, processor,
input, threadNum);
+ return ThreadPoolUtils.sequentialBatchedExecute(
+ executor, processor, input, effectiveThreadNum(threadNum,
executor));
+ }
+
+ /** This method parallel processes one bounded batch and waits for it when
closed. */
+ public static <T, U>
+ ThreadPoolUtils.CloseableBatchIterator<T>
sequentialBatchedExecuteCloseable(
+ Function<U, List<T>> processor, List<U> input, @Nullable
Integer threadNum) {
+ threadNum = normalizeThreadNum(threadNum);
+ ExecutorService executor = getExecutorService(threadNum);
+ return ThreadPoolUtils.sequentialBatchedExecuteCloseable(
+ executor, processor, input, effectiveThreadNum(threadNum,
executor));
}
private static @Nullable Integer normalizeThreadNum(@Nullable Integer
threadNum) {
return threadNum == null || threadNum <= 0 ? null : threadNum;
}
+ private static int effectiveThreadNum(@Nullable Integer threadNum,
ExecutorService executor) {
+ if (threadNum != null) {
+ return threadNum;
+ }
+ return executor instanceof ThreadPoolExecutor
+ ? ((ThreadPoolExecutor) executor).getMaximumPoolSize()
+ : ((SemaphoredDelegatingExecutor) executor).getPermitCount();
+ }
+
/** This method aims to parallel process tasks with randomly but return
values sequentially. */
public static <T, U> Iterator<T> randomlyExecuteSequentialReturn(
Function<U, List<T>> processor, List<U> input, @Nullable Integer
threadNum) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java
index 658507da86..31c6940301 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java
@@ -24,6 +24,7 @@ import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.format.avro.AvroRawBlock;
import org.apache.paimon.fs.Path;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.CollectedDeletes;
@@ -32,6 +33,7 @@ import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.manifest.ManifestAvroReader;
import org.apache.paimon.manifest.ManifestAvroWriter;
+import org.apache.paimon.manifest.ManifestAvroWriter.EncodedBlockMeta;
import org.apache.paimon.manifest.ManifestAvroWriter.EncodedEntry;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
@@ -44,6 +46,7 @@ import org.apache.paimon.stats.StatsTestUtils;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.TraceableFileIO;
import org.junit.jupiter.api.BeforeEach;
@@ -53,6 +56,8 @@ import org.mockito.AdditionalAnswers;
import org.mockito.ArgumentMatchers;
import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.AbstractList;
import java.util.ArrayList;
@@ -60,6 +65,12 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
@@ -68,6 +79,7 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
/** Tests cleanup when manifest rewrites fail with {@link Error}. */
class ManifestRewriteCleanupTest extends ManifestFileMetaTestBase {
@@ -290,6 +302,112 @@ class ManifestRewriteCleanupTest extends
ManifestFileMetaTestBase {
assertNoManifestLeak(manifestCount);
}
+ @Test
+ void testParallelBlockRewriteQuiescesBeforeReleasingDeletes() throws
Exception {
+ ManifestFileMeta first = makeManifest(makeEntry(true, "first"));
+ ManifestFileMeta slow = makeManifest(makeEntry(true, "slow"));
+ List<ManifestFileMeta> input = Arrays.asList(first, slow);
+ int manifestCount = manifestFileCount();
+
+ AssertionError writerFailure = new AssertionError("block writer
failure");
+ RuntimeException planningFailure = new RuntimeException("planning
failure");
+ CountDownLatch slowWorkerStarted = new CountDownLatch(1);
+ CountDownLatch slowWorkerInterrupted = new CountDownLatch(1);
+ CountDownLatch allowSlowWorkerToExit = new CountDownLatch(1);
+ AtomicBoolean slowWorkerExited = new AtomicBoolean();
+ AtomicBoolean deletesReleased = new AtomicBoolean();
+ AtomicBoolean releaseObservedWorkerExit = new AtomicBoolean();
+
+ CollectedDeletes deletes = mock(CollectedDeletes.class);
+ when(deletes.isEmpty()).thenReturn(false);
+ when(deletes.useRowIdFilter()).thenReturn(false);
+ doAnswer(
+ invocation -> {
+ ProjectedManifestEntry entry =
invocation.getArgument(0);
+ if ("first".equals(entry.file().fileName())) {
+ await(slowWorkerStarted);
+ return true;
+ }
+ slowWorkerStarted.countDown();
+ try {
+ awaitIgnoringInterrupts(
+ allowSlowWorkerToExit,
slowWorkerInterrupted);
+ } finally {
+ slowWorkerExited.set(true);
+ }
+ throw planningFailure;
+ })
+ .when(deletes)
+ .copyable(
+ ArgumentMatchers.any(ProjectedManifestEntry.class),
+ ArgumentMatchers.any(ReusableIdentifier.class),
+ ArgumentMatchers.eq(false));
+ doAnswer(
+ invocation -> {
+ deletesReleased.set(true);
+
releaseObservedWorkerExit.set(slowWorkerExited.get());
+ return null;
+ })
+ .when(deletes)
+ .release();
+
+ ManifestFile spyManifestFile = spy(manifestFile);
+ ManifestAvroWriter activeWriter = spy(manifestFile.createAvroWriter());
+ doReturn(activeWriter).when(spyManifestFile).createAvroWriter();
+ doAnswer(
+ invocation -> {
+ activeWriter.write(makeEntry(true, "partial"));
+ throw writerFailure;
+ })
+ .when(activeWriter)
+ .writeEncodedBlock(
+ ArgumentMatchers.any(AvroRawBlock.class),
+ ArgumentMatchers.any(EncodedBlockMeta.class));
+ fileIO.failDeletes();
+
+ ExecutorService caller = Executors.newSingleThreadExecutor();
+ Future<Throwable> rewriteResult =
+ caller.submit(
+ () -> {
+ try {
+ return catchThrowable(
+ () ->
+ invokeBlockRewrite(
+ input,
spyManifestFile, deletes, 2));
+ } finally {
+ deletes.release();
+ }
+ });
+
+ try {
+ assertThat(slowWorkerStarted.await(3, TimeUnit.SECONDS)).isTrue();
+ assertThat(slowWorkerInterrupted.await(3,
TimeUnit.SECONDS)).isTrue();
+ assertThat(slowWorkerExited).isFalse();
+ assertThat(deletesReleased).isFalse();
+ assertThat(rewriteResult.isDone()).isFalse();
+
+ allowSlowWorkerToExit.countDown();
+ Throwable thrown = rewriteResult.get(3, TimeUnit.SECONDS);
+
+ assertThat(thrown).isSameAs(writerFailure);
+ assertThat(slowWorkerExited).isTrue();
+ assertThat(deletesReleased).isTrue();
+ assertThat(releaseObservedWorkerExit).isTrue();
+ assertThat(thrown.getSuppressed())
+ .extracting(Throwable::getMessage)
+ .containsExactly(
+ "Failed to plan manifest rewrite for " +
slow.fileName(),
+ "delete failure 1");
+
assertThat(thrown.getSuppressed()[0].getCause()).isSameAs(planningFailure);
+ assertThat(fileIO.deleteAttempts()).isEqualTo(1);
+ assertNoManifestLeak(manifestCount);
+ } finally {
+ allowSlowWorkerToExit.countDown();
+ caller.shutdownNow();
+ assertThat(caller.awaitTermination(3, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
@Test
void testWriterPreservesWriteErrorAndCleansAllRollingFiles() throws
Exception {
ManifestAvroWriter writer = createManifestFile(1).createAvroWriter();
@@ -388,6 +506,69 @@ class ManifestRewriteCleanupTest extends
ManifestFileMetaTestBase {
assertNoManifestLeak(manifestCount);
}
+ private void invokeBlockRewrite(
+ List<ManifestFileMeta> input,
+ ManifestFile rewriteManifestFile,
+ CollectedDeletes deletes,
+ int parallelism)
+ throws Exception {
+ Method rewriteManifests =
+ ManifestFileBlockMerger.class.getDeclaredMethod(
+ "rewriteManifests",
+ List.class,
+ ManifestFile.class,
+ RowType.class,
+ CollectedDeletes.class,
+ boolean.class,
+ Filter.class,
+ List.class,
+ Integer.class);
+ rewriteManifests.setAccessible(true);
+ try {
+ rewriteManifests.invoke(
+ null,
+ input,
+ rewriteManifestFile,
+ PARTITION_TYPE,
+ deletes,
+ false,
+ null,
+ null,
+ parallelism);
+ } catch (InvocationTargetException e) {
+ Throwable failure = e.getCause();
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ if (failure instanceof Exception) {
+ throw (Exception) failure;
+ }
+ throw new RuntimeException(failure);
+ }
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ if (!latch.await(3, TimeUnit.SECONDS)) {
+ throw new AssertionError("Timed out waiting for the parallel
planning worker.");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static void awaitIgnoringInterrupts(CountDownLatch latch,
CountDownLatch interrupted) {
+ while (true) {
+ try {
+ latch.await();
+ return;
+ } catch (InterruptedException e) {
+ interrupted.countDown();
+ }
+ }
+ }
+
private ManifestEntryRunMergePlan runMergePlan(
ManifestFileMeta input,
CollectedDeletes deletes,
