feat(events): configurable event executor pool sizing + metrics snapshot (#110) (#393)

* feat(events): configurable event executor pool sizing + metrics snapshot (#110)

* test(events): use try-with-resources in #110 executor tests (Sonar S2093)

---------

Co-authored-by: Tomáš Samek <jerry.samek@gmail.com>

Author: tomas-samek

docs/events.md

@@ -107,9 +107,9 @@ The framework dispatches via a bounded `ThreadPoolExecutor` sized for typical sm
 
 | Knob              | Value                                              |
 |-------------------|----------------------------------------------------|
-| Core pool size    | `max(2, cores / 2)`                                |
-| Max pool size     | `cores * 4`                                        |
-| Keep-alive        | 60 seconds                                         |
+| Core pool size    | configurable via `eventExecutorCoreSize(...)` (default `max(2, cores / 2)`) |
+| Max pool size     | configurable via `eventExecutorMaxSize(...)` (default `cores * 4`)  |
+| Keep-alive        | configurable via `eventExecutorKeepAlive(...)` (default 60 seconds) |
 | Queue             | bounded `LinkedBlockingQueue`, capacity `queueCapacity` (default 1024) |
 | Rejection policy  | chosen by `onOverflow(...)` (default `CALLER_RUNS` — slows publisher under overload) |
 | Thread name       | `tiko-event-async-{n}` (daemon)                    |
@@ -134,6 +134,31 @@ TikoOptions opts = TikoOptions.builder()
 
 Whatever the policy, once the container is shutting down a rejected task degrades to an observable logged drop — `BLOCK` never hangs teardown and `THROW` never throws on the shutdown path.
 
+#### Tuning the default executor's pool size
+
+Size the pool for your workload — more threads for I/O-bound handlers, fewer for CPU-bound. Omitting a knob keeps its processor-derived default, so these are purely additive:
+
+```java
+TikoOptions opts = TikoOptions.builder()
+        .eventExecutorCoreSize(8)                    // default max(2, cores / 2)
+        .eventExecutorMaxSize(32)                    // default cores * 4
+        .eventExecutorKeepAlive(Duration.ofSeconds(45)) // default 60s
+        .build();
+```
+
+The effective max size must be `>=` the effective core size, or container start fails with `IllegalArgumentException`.
+
+#### Observing the default executor
+
+Poll a point-in-time snapshot of the framework executor for monitoring (saturation, queue depth, throughput):
+
+```java
+container.eventExecutorMetrics().ifPresent(m ->
+        meterRegistry.gauge("tiko.events.queue", m.queueSize()));
+```
+
+`eventExecutorMetrics()` returns `Optional<ExecutorMetrics>` — a record exposing `activeCount`, `poolSize`, `queueSize`, `queueRemainingCapacity`, `completedTaskCount`, `corePoolSize`, and `maxPoolSize`. It is **empty** when you supply your own executor (its metrics are yours to expose, not the framework's).
+
 Workloads with extreme throughput or latency requirements can supply their own executor instead:
 
 ```java
@@ -144,7 +169,7 @@ TikoOptions opts = TikoOptions.builder()
 try (Container container = Tiko.create(opts)) { ... }
 ```
 
-When you supply your own executor, **you own its lifecycle** — `Container.shutdown()` does not stop it, and `queueCapacity` / `onOverflow` have no effect (your executor brings its own queue and rejection policy). Async handler exceptions still route to the configured `ErrorHandler` regardless of which executor is in use.
+When you supply your own executor, **you own its lifecycle** — `Container.shutdown()` does not stop it; `queueCapacity` / `onOverflow` and the pool-size knobs have no effect (your executor brings its own queue, rejection policy, and sizing); and `eventExecutorMetrics()` returns empty. Async handler exceptions still route to the configured `ErrorHandler` regardless of which executor is in use.
 
 ### Execution timeouts
 

tiko-api/src/main/java/io/tiko/Container.java

@@ -191,6 +191,22 @@ default EventBus events() {
      */
     java.util.concurrent.ExecutorService getEventExecutor();
 
+    /**
+     * Returns a point-in-time snapshot of the framework's event-dispatch executor (#110),
+     * for monitoring saturation and queue depth.
+     *
+     * <p>Returns {@link java.util.Optional#empty()} when the container does not own a
+     * sampleable executor — chiefly when a custom executor was supplied via
+     * {@code TikoOptions.eventExecutor(...)} (its lifecycle and metrics are yours, not the
+     * framework's). The default implementation returns {@code empty()} so user-supplied
+     * {@code Container} implementations remain source-compatible.
+     *
+     * @return a snapshot of the framework executor, or empty when none is owned
+     */
+    default java.util.Optional<ExecutorMetrics> eventExecutorMetrics() {
+        return java.util.Optional.empty();
+    }
+
     /**
      * Returns the error handler configured for this container.
      *

tiko-api/src/main/java/io/tiko/ExecutorMetrics.java

@@ -0,0 +1,48 @@
+package io.tiko;
+
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * An immutable point-in-time snapshot of the framework's event-dispatch executor (#110).
+ *
+ * <p>Obtained via {@link Container#eventExecutorMetrics()}. The values mirror the
+ * corresponding {@link ThreadPoolExecutor} getters at the instant the snapshot was taken;
+ * because the pool is live and concurrent, they are a consistent-enough sample for
+ * monitoring (saturation, queue depth, throughput), not a transactional view.
+ *
+ * @param activeCount            threads actively executing tasks ({@link ThreadPoolExecutor#getActiveCount()})
+ * @param poolSize               current number of threads in the pool ({@link ThreadPoolExecutor#getPoolSize()})
+ * @param queueSize              tasks waiting in the work queue
+ * @param queueRemainingCapacity free slots left in the bounded work queue before overflow
+ * @param completedTaskCount     approximate count of tasks that have completed
+ *                               ({@link ThreadPoolExecutor#getCompletedTaskCount()})
+ * @param corePoolSize           configured core pool size ({@link ThreadPoolExecutor#getCorePoolSize()})
+ * @param maxPoolSize            configured maximum pool size ({@link ThreadPoolExecutor#getMaximumPoolSize()})
+ */
+public record ExecutorMetrics(
+        int activeCount,
+        int poolSize,
+        int queueSize,
+        int queueRemainingCapacity,
+        long completedTaskCount,
+        int corePoolSize,
+        int maxPoolSize) {
+
+    /**
+     * Samples the given executor into an immutable snapshot.
+     *
+     * @param tpe the live framework executor to sample
+     * @return a snapshot of {@code tpe}'s current state
+     */
+    public static ExecutorMetrics from(ThreadPoolExecutor tpe) {
+        var queue = tpe.getQueue();
+        return new ExecutorMetrics(
+                tpe.getActiveCount(),
+                tpe.getPoolSize(),
+                queue.size(),
+                queue.remainingCapacity(),
+                tpe.getCompletedTaskCount(),
+                tpe.getCorePoolSize(),
+                tpe.getMaximumPoolSize());
+    }
+}

tiko-api/src/test/java/io/tiko/ExecutorMetricsTest.java

@@ -0,0 +1,59 @@
+package io.tiko;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+
+class ExecutorMetricsTest {
+
+    @Test
+    void fromMirrorsThreadPoolExecutorState() {
+        try (ThreadPoolExecutor tpe =
+                new ThreadPoolExecutor(2, 6, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(16))) {
+            ExecutorMetrics m = ExecutorMetrics.from(tpe);
+
+            assertThat(m.corePoolSize()).isEqualTo(2);
+            assertThat(m.maxPoolSize()).isEqualTo(6);
+            assertThat(m.queueSize()).isZero();
+            assertThat(m.queueRemainingCapacity()).isEqualTo(16);
+            assertThat(m.activeCount()).isZero();
+            assertThat(m.completedTaskCount()).isZero();
+        }
+    }
+
+    @Test
+    void fromReflectsQueuedAndCompletedWork() throws Exception {
+        // Single thread, capacity-1 queue: one task occupies the worker, the rest sit in the queue.
+        ThreadPoolExecutor tpe = new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(8));
+        var gate = new java.util.concurrent.CountDownLatch(1);
+        var started = new java.util.concurrent.CountDownLatch(1);
+        try {
+            tpe.execute(() -> {
+                started.countDown();
+                try {
+                    gate.await();
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+            });
+            assertThat(started.await(2, TimeUnit.SECONDS)).isTrue();
+            tpe.execute(() -> {}); // queued behind the blocked worker
+
+            ExecutorMetrics busy = ExecutorMetrics.from(tpe);
+            assertThat(busy.activeCount()).isEqualTo(1);
+            assertThat(busy.queueSize()).isEqualTo(1);
+
+            gate.countDown();
+            tpe.shutdown();
+            assertThat(tpe.awaitTermination(2, TimeUnit.SECONDS)).isTrue();
+
+            ExecutorMetrics done = ExecutorMetrics.from(tpe);
+            assertThat(done.completedTaskCount()).isEqualTo(2);
+        } finally {
+            tpe.shutdownNow();
+        }
+    }
+}

tiko-runtime/src/main/java/io/tiko/runtime/AggregatingContainer.java

@@ -149,7 +149,12 @@ public AggregatingContainer(
         this.errorHandler = errorHandler;
         this.eventExecutor = userEventExecutor != null
                 ? userEventExecutor
-                : DefaultEventExecutorFactory.create(options.queueCapacity(), options.onOverflow());
+                : DefaultEventExecutorFactory.create(
+                        options.queueCapacity(),
+                        options.onOverflow(),
+                        options.eventExecutorCoreSize(),
+                        options.eventExecutorMaxSize(),
+                        options.eventExecutorKeepAlive());
         this.ownsEventExecutor = (userEventExecutor == null);
         this.shutdownTimeout = shutdownTimeout;
         this.options = options;
@@ -498,6 +503,15 @@ public java.util.concurrent.ExecutorService getEventExecutor() {
         return eventExecutor;
     }
 
+    @Override
+    public java.util.Optional<io.tiko.ExecutorMetrics> eventExecutorMetrics() {
+        // Only the framework-owned default pool is sampleable (#110). A user-supplied executor —
+        // even a ThreadPoolExecutor — is the user's to observe; we report empty for it.
+        return ownsEventExecutor && eventExecutor instanceof java.util.concurrent.ThreadPoolExecutor tpe
+                ? java.util.Optional.of(io.tiko.ExecutorMetrics.from(tpe))
+                : java.util.Optional.empty();
+    }
+
     @Override
     public void runInEventScope(Runnable task) {
         runNested(moduleContainers.iterator(), wrapWithUnitLifecycle(task), Container::runInEventScope);

tiko-runtime/src/main/java/io/tiko/runtime/DefaultEventExecutorFactory.java

@@ -1,5 +1,6 @@
 package io.tiko.runtime;
 
+import java.time.Duration;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.RejectedExecutionHandler;
@@ -14,9 +15,9 @@
  *
  * <p>Configuration:
  * <ul>
- *   <li>Core pool: {@code Math.max(2, availableProcessors() / 2)}</li>
- *   <li>Max pool: {@code availableProcessors() * 4}</li>
- *   <li>Keep-alive: 60 seconds</li>
+ *   <li>Core pool: configurable (default {@code Math.max(2, availableProcessors() / 2)}, #110)</li>
+ *   <li>Max pool: configurable (default {@code availableProcessors() * 4}, #110)</li>
+ *   <li>Keep-alive: configurable (default 60 seconds, #110)</li>
  *   <li>Queue: bounded {@link LinkedBlockingQueue}, capacity configurable (default 1024, #109)</li>
  *   <li>Rejection policy: chosen by the {@link OverflowPolicy} (default {@code CALLER_RUNS}, #109).
  *       Every policy degrades to an observable WARNING (never a silent drop, never a hang, never a
@@ -32,6 +33,9 @@ public final class DefaultEventExecutorFactory {
     /** Historical defaults, applied when callers do not configure backpressure (#109). */
     static final int DEFAULT_QUEUE_CAPACITY = 1024;
 
+    /** Default non-core thread idle keep-alive (#110). */
+    static final long DEFAULT_KEEP_ALIVE_SECONDS = 60L;
+
     private DefaultEventExecutorFactory() {}
 
     public static ExecutorService create() {
@@ -40,12 +44,33 @@ public static ExecutorService create() {
 
     /**
      * Builds the default event executor with a bounded queue of {@code queueCapacity} and the
-     * given {@code overflowPolicy} for tasks rejected when that queue is full (#109).
+     * given {@code overflowPolicy} for tasks rejected when that queue is full (#109), using the
+     * framework's processor-derived pool sizing.
      */
     public static ExecutorService create(int queueCapacity, OverflowPolicy overflowPolicy) {
+        return create(queueCapacity, overflowPolicy, TikoOptions.UNSET_POOL_SIZE, TikoOptions.UNSET_POOL_SIZE, null);
+    }
+
+    /**
+     * Builds the default event executor with explicit pool sizing (#110). A {@link
+     * TikoOptions#UNSET_POOL_SIZE} core or max size, or a {@code null} keep-alive, falls back to
+     * the framework's processor-derived defaults (core {@code Math.max(2, cores / 2)}, max {@code
+     * cores * 4}, keep-alive {@value #DEFAULT_KEEP_ALIVE_SECONDS}s) — so omitting every #110 knob
+     * reproduces the historical behaviour exactly.
+     *
+     * @throws IllegalArgumentException if the effective max pool size is less than the effective core size
+     */
+    public static ExecutorService create(
+            int queueCapacity, OverflowPolicy overflowPolicy, int coreSize, int maxSize, Duration keepAlive) {
         int cores = Runtime.getRuntime().availableProcessors();
-        int corePoolSize = Math.max(2, cores / 2);
-        int maxPoolSize = cores * 4;
+        int corePoolSize = coreSize == TikoOptions.UNSET_POOL_SIZE ? Math.max(2, cores / 2) : coreSize;
+        int maxPoolSize = maxSize == TikoOptions.UNSET_POOL_SIZE ? cores * 4 : maxSize;
+        if (maxPoolSize < corePoolSize) {
+            throw new IllegalArgumentException("eventExecutorMaxSize (" + maxPoolSize
+                    + ") must be >= eventExecutorCoreSize (" + corePoolSize + ")");
+        }
+        long keepAliveMs =
+                keepAlive == null ? TimeUnit.SECONDS.toMillis(DEFAULT_KEEP_ALIVE_SECONDS) : keepAlive.toMillis();
 
         ThreadFactory threadFactory = new ThreadFactory() {
             private final AtomicInteger counter = new AtomicInteger();
@@ -61,8 +86,8 @@ public Thread newThread(Runnable r) {
         return new ThreadPoolExecutor(
                 corePoolSize,
                 maxPoolSize,
-                60L,
-                TimeUnit.SECONDS,
+                keepAliveMs,
+                TimeUnit.MILLISECONDS,
                 new LinkedBlockingQueue<>(queueCapacity),
                 threadFactory,
                 rejectionHandlerFor(overflowPolicy));