BewareMyPower commented on code in PR #25187:
URL: https://github.com/apache/pulsar/pull/25187#discussion_r2751260385
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java:
##########
@@ -616,30 +623,30 @@ public void invalidateCaches(String...paths) {
}
}
- /**
- * Run the task in the executor thread and fail the future if the executor
is shutting down.
- */
- @VisibleForTesting
- public void execute(Runnable task, CompletableFuture<?> future) {
+ protected final <T> void processEvent(Consumer<T> eventProcessor, T event)
{
try {
- executor.execute(task);
- } catch (Throwable t) {
- future.completeExceptionally(t);
+ eventExecutor.execute(() -> eventProcessor.accept(event));
+ } catch (RejectedExecutionException e) {
+ log.warn("Rejected processing event {}", event);
}
}
- /**
- * Run the task in the executor thread and fail the future if the executor
is shutting down.
- */
- @VisibleForTesting
- public void execute(Runnable task, Supplier<List<CompletableFuture<?>>>
futures) {
+ protected final void scheduleDelayedTask(long delay, TimeUnit unit,
Runnable task) {
+ CompletableFuture.delayedExecutor(delay, unit).execute(task);
Review Comment:
Let me try a dedicated scheduler pool for such retries, mark this PR as
drafted.
--
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]
