You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@polaris.apache.org by sn...@apache.org on 2026/05/29 05:59:42 UTC
(polaris) branch main updated: NoSQL: async exec: handle rejected-execution (#4543)
This is an automated email from the ASF dual-hosted git repository.
snazy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git
The following commit(s) were added to refs/heads/main by this push:
new fb4e5c0f53 NoSQL: async exec: handle rejected-execution (#4543)
fb4e5c0f53 is described below
commit fb4e5c0f53e13377fd7b70519ed604af9587f539
Author: Robert Stupp <sn...@snazy.de>
AuthorDate: Fri May 29 07:59:37 2026 +0200
NoSQL: async exec: handle rejected-execution (#4543)
The async executors can briefly reject work when the backing pool is saturated, especially with the SynchronousQueue based Vert.x executor.
Treat that as temporary pressure instead of failing the task immediately.
Rejected submissions are rescheduled after a short delay, unless the task was already cancelled or the executor is shutting down.
Cancellation also clears the retry timer so we do not run cancelled work later.
This is hardening work, not an immediate fix, because currently there are maybe a handful of tasks scheduled at once.
---
.../nosql/async/java/JavaPoolAsyncExec.java | 19 +++++++-
.../nosql/async/java/TestJavaPoolAsyncExec.java | 9 ++--
persistence/nosql/async/vertx/build.gradle.kts | 3 ++
.../polaris/nosql/async/vertx/VertxAsyncExec.java | 54 ++++++++++++++++++++++
.../nosql/async/vertx/TestVertxAsyncExec.java | 14 ++----
5 files changed, 83 insertions(+), 16 deletions(-)
diff --git a/persistence/nosql/async/java/src/main/java/org/apache/polaris/nosql/async/java/JavaPoolAsyncExec.java b/persistence/nosql/async/java/src/main/java/org/apache/polaris/nosql/async/java/JavaPoolAsyncExec.java
index 40d8717027..2d62a606f2 100644
--- a/persistence/nosql/async/java/src/main/java/org/apache/polaris/nosql/async/java/JavaPoolAsyncExec.java
+++ b/persistence/nosql/async/java/src/main/java/org/apache/polaris/nosql/async/java/JavaPoolAsyncExec.java
@@ -63,6 +63,7 @@ import org.slf4j.MDC;
public class JavaPoolAsyncExec implements AsyncExec, AutoCloseable {
private static final Logger LOGGER = LoggerFactory.getLogger(JavaPoolAsyncExec.class.getName());
private static final Duration MAX_DURATION = Duration.ofDays(7);
+ private static final long SUBMISSION_RETRY_MILLIS = 10L;
public static final String EXECUTOR_THREAD_NAME_PREFIX = "JavaPoolTaskExecutor#";
public static final String SCHEDULER_THREAD_NAME_PREFIX = "JavaPoolTaskScheduler#";
@@ -239,7 +240,18 @@ public class JavaPoolAsyncExec implements AsyncExec, AutoCloseable {
@SuppressWarnings("FutureReturnValueIgnored")
private void immediate(CancelableFuture<?> cancelable) {
- executorService.submit(cancelable);
+ try {
+ executorService.submit(cancelable);
+ } catch (RejectedExecutionException e) {
+ // In case the pool is being shut-down, do not attempt to reschedule.
+ if (!cancelable.cancelledOrShutdown()) {
+ try {
+ delayed(cancelable, SUBMISSION_RETRY_MILLIS);
+ } catch (RejectedExecutionException retryRejected) {
+ cancelable.completeExceptionally(retryRejected);
+ }
+ }
+ }
}
@SuppressWarnings("FutureReturnValueIgnored")
@@ -341,6 +353,11 @@ public class JavaPoolAsyncExec implements AsyncExec, AutoCloseable {
return completable.isCancelled() || shutdown;
}
+ private void completeExceptionally(Throwable t) {
+ completable.completeExceptionally(t);
+ tasks.remove(this);
+ }
+
@Override
public CompletionStage<R> completionStage() {
return completable;
diff --git a/persistence/nosql/async/java/src/test/java/org/apache/polaris/nosql/async/java/TestJavaPoolAsyncExec.java b/persistence/nosql/async/java/src/test/java/org/apache/polaris/nosql/async/java/TestJavaPoolAsyncExec.java
index fd3f797bb9..23dfee8bf8 100644
--- a/persistence/nosql/async/java/src/test/java/org/apache/polaris/nosql/async/java/TestJavaPoolAsyncExec.java
+++ b/persistence/nosql/async/java/src/test/java/org/apache/polaris/nosql/async/java/TestJavaPoolAsyncExec.java
@@ -20,11 +20,9 @@ package org.apache.polaris.nosql.async.java;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
@@ -129,7 +127,7 @@ public class TestJavaPoolAsyncExec extends AsyncExecTestBase {
}
@Test
- public void immediateTaskRejectedWhenThreadPoolIsSaturated() throws Exception {
+ public void immediateTaskRetriesWhenThreadPoolIsSaturated() throws Exception {
var started = new CountDownLatch(1);
var release = new Semaphore(0);
@@ -143,11 +141,12 @@ public class TestJavaPoolAsyncExec extends AsyncExecTestBase {
});
assertThat(started.await(5_000, MILLISECONDS)).isTrue();
- assertThatThrownBy(() -> executor.submit(() -> "rejected"))
- .isInstanceOf(RejectedExecutionException.class);
+ var delayed = executor.submit(() -> "delayed").completionStage().toCompletableFuture();
+ assertThat(delayed).isNotDone();
release.release();
assertThat(blocker.completionStage()).succeedsWithin(Duration.ofSeconds(5));
+ assertThat(delayed).succeedsWithin(Duration.ofSeconds(5)).isEqualTo("delayed");
}
}
}
diff --git a/persistence/nosql/async/vertx/build.gradle.kts b/persistence/nosql/async/vertx/build.gradle.kts
index 2bb080a48e..69893b9471 100644
--- a/persistence/nosql/async/vertx/build.gradle.kts
+++ b/persistence/nosql/async/vertx/build.gradle.kts
@@ -52,6 +52,9 @@ dependencies {
testFixturesApi("io.vertx:vertx-core")
testImplementation(testFixtures(project(":polaris-async-api")))
+
+ testCompileOnly(platform(libs.jackson.bom))
+ testCompileOnly("com.fasterxml.jackson.core:jackson-databind")
}
tasks.withType<Javadoc> {
diff --git a/persistence/nosql/async/vertx/src/main/java/org/apache/polaris/nosql/async/vertx/VertxAsyncExec.java b/persistence/nosql/async/vertx/src/main/java/org/apache/polaris/nosql/async/vertx/VertxAsyncExec.java
index e5ce6dbb2e..8e8f102012 100644
--- a/persistence/nosql/async/vertx/src/main/java/org/apache/polaris/nosql/async/vertx/VertxAsyncExec.java
+++ b/persistence/nosql/async/vertx/src/main/java/org/apache/polaris/nosql/async/vertx/VertxAsyncExec.java
@@ -60,6 +60,7 @@ import org.slf4j.MDC;
class VertxAsyncExec implements AsyncExec, AutoCloseable {
private static final Logger LOGGER = LoggerFactory.getLogger(VertxAsyncExec.class.getName());
private static final Duration MAX_DURATION = Duration.ofDays(7);
+ private static final long SUBMISSION_RETRY_MILLIS = 10L;
public static final String EXECUTOR_THREAD_NAME_PREFIX = "VertxAsyncExec#";
@@ -203,6 +204,8 @@ class VertxAsyncExec implements AsyncExec, AutoCloseable {
private final AtomicReference<TimerState> timerState =
new AtomicReference<>(NotStarted.INSTANCE);
+ private final AtomicReference<TimerState> retryTimerState =
+ new AtomicReference<>(NotStarted.INSTANCE);
private final CompletableFuture<R> completable = new CompletableFuture<>();
private final Runnable runnable;
@@ -264,12 +267,38 @@ class VertxAsyncExec implements AsyncExec, AutoCloseable {
if (periodic && !running.compareAndSet(false, true)) {
return;
}
+ submitAsync();
+ }
+
+ @SuppressWarnings("FutureReturnValueIgnored")
+ private void submitAsync() {
try {
var f = executorService.submit(this);
runningFuture.set(f);
VertxAsyncExec.this.taskSubmittedHook(f);
+ } catch (Throwable e) {
+ if (cancelledOrShutdown()) {
+ if (periodic) {
+ running.set(false);
+ }
+ cancel();
+ } else {
+ scheduleSubmissionRetry();
+ }
+ }
+ }
+
+ private void scheduleSubmissionRetry() {
+ try {
+ var timerId = vertx.setTimer(SUBMISSION_RETRY_MILLIS, this::execAsyncRetry);
+ if (!retryTimerState.compareAndSet(NotStarted.INSTANCE, new Started(timerId))) {
+ vertx.cancelTimer(timerId);
+ } else if (cancelledOrShutdown()) {
+ cancelRetryTimer();
+ }
} catch (Throwable e) {
cancelTimer();
+ cancelRetryTimer();
runningFuture.set(null);
if (periodic) {
running.set(false);
@@ -279,6 +308,23 @@ class VertxAsyncExec implements AsyncExec, AutoCloseable {
}
}
+ private void execAsyncRetry(long unused) {
+ var previous =
+ retryTimerState.getAndUpdate(
+ state -> state instanceof Canceled ? state : NotStarted.INSTANCE);
+ if (previous instanceof Canceled) {
+ return;
+ }
+ if (cancelledOrShutdown()) {
+ if (periodic) {
+ running.set(false);
+ }
+ cancel();
+ } else {
+ submitAsync();
+ }
+ }
+
@Override
public void run() {
VertxAsyncExec.this.taskRunHook();
@@ -314,6 +360,13 @@ class VertxAsyncExec implements AsyncExec, AutoCloseable {
}
}
+ private void cancelRetryTimer() {
+ var previous = retryTimerState.getAndSet(Canceled.INSTANCE);
+ if (previous instanceof Started(long timerId)) {
+ vertx.cancelTimer(timerId);
+ }
+ }
+
private boolean cancelledOrShutdown() {
return completable.isCancelled() || shutdown;
}
@@ -326,6 +379,7 @@ class VertxAsyncExec implements AsyncExec, AutoCloseable {
@Override
public void cancel() {
cancelTimer();
+ cancelRetryTimer();
completable.cancel(false);
var f = runningFuture.getAndSet(null);
if (f != null && !f.isDone()) {
diff --git a/persistence/nosql/async/vertx/src/test/java/org/apache/polaris/nosql/async/vertx/TestVertxAsyncExec.java b/persistence/nosql/async/vertx/src/test/java/org/apache/polaris/nosql/async/vertx/TestVertxAsyncExec.java
index 6a42242020..a996995a66 100644
--- a/persistence/nosql/async/vertx/src/test/java/org/apache/polaris/nosql/async/vertx/TestVertxAsyncExec.java
+++ b/persistence/nosql/async/vertx/src/test/java/org/apache/polaris/nosql/async/vertx/TestVertxAsyncExec.java
@@ -26,9 +26,7 @@ import io.vertx.core.Vertx;
import jakarta.inject.Inject;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
-import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -132,7 +130,7 @@ public class TestVertxAsyncExec extends AsyncExecTestBase {
}
@Test
- public void immediateTaskFailsWhenThreadPoolIsSaturated() throws Exception {
+ public void immediateTaskRetriesWhenThreadPoolIsSaturated() throws Exception {
var started = new CountDownLatch(1);
var release = new Semaphore(0);
@@ -147,16 +145,12 @@ public class TestVertxAsyncExec extends AsyncExecTestBase {
});
assertThat(started.await(5, SECONDS)).isTrue();
- var rejected = executor.submit(() -> "rejected").completionStage().toCompletableFuture();
- assertThat(rejected)
- .failsWithin(Duration.ofSeconds(5))
- .withThrowableThat()
- .isInstanceOf(ExecutionException.class)
- .havingCause()
- .isInstanceOf(RejectedExecutionException.class);
+ var delayed = executor.submit(() -> "delayed").completionStage().toCompletableFuture();
+ assertThat(delayed).isNotDone();
release.release();
assertThat(blocker.completionStage()).succeedsWithin(Duration.ofSeconds(5));
+ assertThat(delayed).succeedsWithin(Duration.ofSeconds(5)).isEqualTo("delayed");
}
}
}