JingsongLi commented on code in PR #9397:
URL: https://github.com/apache/paimon/pull/9397#discussion_r3869610545
##########
paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java:
##########
@@ -280,6 +435,7 @@ private void reportPartitions(
}
// A commit that replaced what the partitions held reports a total; an
appending one saw
// only its own files, so its numbers are an increment.
+ markPublishedTargetsToPreserveOnAbort(messages);
Review Comment:
[P1] Do not preserve published append files for every metadata failure
This flag is set before `createPartitions`, so even a definite pre-mutation
failure (for example, loading the catalog or an authorization rejection) makes
`abort` skip deletion for every published target. For an append to an already
registered partition, scans list files directly from the partition directory,
so the failed write remains visible and a retry can duplicate the rows. The new
partial-batch test also preserves targets from the batch that throws before
applying anything. Please track outcomes per request or partition: roll back
definitely unapplied targets, and reconcile or retry idempotently only when the
metadata outcome is genuinely indeterminate.
##########
paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java:
##########
@@ -263,32 +322,53 @@ 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();
+ continue;
+ }
+ if (mode.slidingWindow || activeTasks.isEmpty()) {
+ fillWindow();
+ }
+ if (activeTasks.isEmpty()) {
+ return;
+ }
+ 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();
- activeResults = results.iterator();
- } catch (RuntimeException | Error failure) {
- if (task.failureReported()) {
- activeTasks.poll();
- }
- throw failure;
}
- } else if (batches.isEmpty()) {
- return;
- } else {
- submitBatch(batches.poll());
+ throw failure;
}
}
}
- private void submitBatch(List<U> batch) {
+ /** Does not consume more input than the active-task window can hold.
*/
+ private void fillWindow() {
ClassLoader classLoader =
Thread.currentThread().getContextClassLoader();
- for (U input : batch) {
- BatchTask<T, U> task = new BatchTask<>(processor, input,
classLoader);
- executor.execute(task);
- activeTasks.add(task);
+ AccessControlContext accessControlContext =
AccessController.getContext();
+ while (activeTasks.size() < queueSize) {
+ synchronized (submissionLock) {
+ if (submissionStopped || !input.hasNext()) {
+ return;
+ }
+ BatchTask<T, U> task =
+ new BatchTask<>(
+ processor,
+ input.next(),
+ classLoader,
+ accessControlContext,
+ this::stopSubmission);
+ executor.execute(task);
Review Comment:
[P2] Avoid calling a potentially blocking executor while holding
`submissionLock`
`SemaphoredDelegatingExecutor.execute` blocks while acquiring a permit. With
`permitCount = 1` and `queueSize = 2`, the second submission holds
`submissionLock` while waiting for the only permit; if the first task fails,
`stopSubmission` waits for the same lock, so that task cannot return and
release its permit. The iterator and close path then deadlock. Please make
failure signaling lock-free or move `execute` outside any lock required by
workers, and add a timed permit-1/window-2 regression test.
##########
paimon-api/src/main/java/org/apache/paimon/utils/ThreadPoolUtils.java:
##########
@@ -372,35 +469,55 @@ public void run() {
runner = Thread.currentThread();
}
+ Thread currentThread = Thread.currentThread();
+ boolean interruptedOnEntry = currentThread.isInterrupted();
+ ClassLoader originalClassLoader =
currentThread.getContextClassLoader();
try {
- Thread.currentThread().setContextClassLoader(classLoader);
- result = processor.apply(input);
+ currentThread.setContextClassLoader(classLoader);
+ result =
+ AccessController.doPrivileged(
+ (PrivilegedAction<List<T>>) () ->
processor.apply(input),
+ accessControlContext);
} catch (RuntimeException | Error taskFailure) {
failure = taskFailure;
+ stopSubmission.run();
} finally {
+ currentThread.setContextClassLoader(originalClassLoader);
Review Comment:
[P2] Always publish task completion if TCCL handling fails
Under a JDK 8 `SecurityManager` that denies
`RuntimePermission("setContextClassLoader")`, line 476 records the first
`SecurityException`, but this restore throws again before `state = FINISHED`
and `completion.countDown()`, leaving `result()` and `close()` blocked forever.
`getContextClassLoader()` at line 474 is also outside the protected region.
Please put state publication and latch countdown in an outermost `finally`,
handle TCCL restore failures separately, and add a timed
denying-`SecurityManager` regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
