sundapeng commented on code in PR #9397:
URL: https://github.com/apache/paimon/pull/9397#discussion_r3874716273
##########
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:
Fixed in 9372cc5c05. Removed submissionLock, made failure signaling volatile
and lock-free, and no longer invokes a potentially blocking executor while
holding a lock needed by workers. Added a timed permit=1/window=2
SemaphoredDelegatingExecutor 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:
Fixed in 9372cc5c05. Context-class-loader get, set, and restore failures are
handled explicitly; restore failures become primary or suppressed as
appropriate; and an outermost finally always publishes FINISHED and counts down
the completion latch. Added timed tests for get denial, restore-only failure,
set-plus-restore dual failure, and a null original context class loader on JDK
8.
##########
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:
Fixed in 9372cc5c05. Append now performs idempotent partition registration
before additive statistics, rolls back all unique files from this attempt when
catalog or Hive registration fails (possibly leaving only harmless empty
partition entries), marks the serialized commit messages only after all
registrations succeed, and makes additive statistics one-shot and best-effort
so an indeterminate statistics response cannot replay the data commit. Added
coverage for catalog-loader failure, applied-then-lost registration responses,
partial batch failures, Hive partial registration, serialized abort state, and
statistics response loss. Strict exactly-once additive statistics still
requires server-side idempotency.
--
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]
