CAMEL-20199: Add JMX support for virtual thread executors (#21701)

Virtual thread executors (ThreadPerTaskExecutor) were silently skipped
from JMX registration because LifecycleStrategy.onThreadPoolAdd only
accepted ThreadPoolExecutor. This adds overloaded methods for
ExecutorService to register non-ThreadPoolExecutor instances in JMX.

New ManagedVirtualThreadExecutorMBean exposes basic attributes: id,
sourceId, routeId, isVirtualThread, and isShutdown.

Author: gnodet

core/camel-api/src/main/java/org/apache/camel/spi/LifecycleStrategy.java

@@ -17,6 +17,7 @@
 package org.apache.camel.spi;
 
 import java.util.Collection;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import org.apache.camel.CamelContext;
@@ -209,4 +210,30 @@ void onThreadPoolAdd(
      */
     void onThreadPoolRemove(CamelContext camelContext, ThreadPoolExecutor threadPool);
 
+    /**
+     * Notification on adding an executor service (such as a virtual thread executor) that is not a
+     * {@link ThreadPoolExecutor}.
+     *
+     * @param camelContext        the camel context
+     * @param executorService     the executor service
+     * @param id                  id of the thread pool (can be null in special cases)
+     * @param sourceId            id of the source creating the thread pool (can be null in special cases)
+     * @param routeId             id of the route for the source (is null if no source)
+     * @param threadPoolProfileId id of the thread pool profile, if used for creating this thread pool (can be null)
+     */
+    default void onThreadPoolAdd(
+            CamelContext camelContext, ExecutorService executorService, String id,
+            String sourceId, String routeId, String threadPoolProfileId) {
+    }
+
+    /**
+     * Notification on removing an executor service (such as a virtual thread executor) that is not a
+     * {@link ThreadPoolExecutor}.
+     *
+     * @param camelContext    the camel context
+     * @param executorService the executor service
+     */
+    default void onThreadPoolRemove(CamelContext camelContext, ExecutorService executorService) {
+    }
+
 }

core/camel-api/src/main/java/org/apache/camel/spi/ManagementObjectNameStrategy.java

@@ -16,6 +16,7 @@
  */
 package org.apache.camel.spi;
 
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import javax.management.MalformedObjectNameException;
@@ -78,6 +79,12 @@ ObjectName getObjectNameForClusterService(CamelContext context, CamelClusterServ
     ObjectName getObjectNameForThreadPool(CamelContext context, ThreadPoolExecutor threadPool, String id, String sourceId)
             throws MalformedObjectNameException;
 
+    default ObjectName getObjectNameForThreadPool(
+            CamelContext context, ExecutorService executorService, String id, String sourceId)
+            throws MalformedObjectNameException {
+        return null;
+    }
+
     ObjectName getObjectNameForEventNotifier(CamelContext context, EventNotifier eventNotifier)
             throws MalformedObjectNameException;
 }

core/camel-api/src/main/java/org/apache/camel/spi/ManagementObjectStrategy.java

@@ -16,6 +16,7 @@
  */
 package org.apache.camel.spi;
 
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import org.apache.camel.CamelContext;
@@ -67,5 +68,11 @@ Object getManagedObjectForThreadPool(
             CamelContext context, ThreadPoolExecutor threadPool,
             String id, String sourceId, String routeId, String threadPoolProfileId);
 
+    default Object getManagedObjectForThreadPool(
+            CamelContext context, ExecutorService executorService,
+            String id, String sourceId, String routeId, String threadPoolProfileId) {
+        return null;
+    }
+
     Object getManagedObjectForEventNotifier(CamelContext context, EventNotifier eventNotifier);
 }

core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/BaseExecutorServiceManager.java

@@ -372,6 +372,11 @@ private void doRemove(ExecutorService executorService, boolean failSafe) {
             for (LifecycleStrategy lifecycle : camelContext.getLifecycleStrategies()) {
                 lifecycle.onThreadPoolRemove(camelContext, threadPool);
             }
+        } else {
+            // for non-ThreadPoolExecutor instances (e.g., virtual thread executors)
+            for (LifecycleStrategy lifecycle : camelContext.getLifecycleStrategies()) {
+                lifecycle.onThreadPoolRemove(camelContext, executorService);
+            }
         }
 
         // remove reference as its shutdown (do not remove if fail-safe)
@@ -592,6 +597,11 @@ private void onThreadPoolCreated(ExecutorService executorService, Object source,
             for (LifecycleStrategy lifecycle : camelContext.getLifecycleStrategies()) {
                 lifecycle.onThreadPoolAdd(camelContext, threadPool, id, sourceId, routeId, threadPoolProfileId);
             }
+        } else {
+            // for non-ThreadPoolExecutor instances (e.g., virtual thread executors)
+            for (LifecycleStrategy lifecycle : camelContext.getLifecycleStrategies()) {
+                lifecycle.onThreadPoolAdd(camelContext, executorService, id, sourceId, routeId, threadPoolProfileId);
+            }
         }
 
         // now call strategy to allow custom logic

core/camel-management-api/src/main/java/org/apache/camel/api/management/mbean/ManagedVirtualThreadExecutorMBean.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.api.management.mbean;
+
+import org.apache.camel.api.management.ManagedAttribute;
+
+public interface ManagedVirtualThreadExecutorMBean {
+
+    @ManagedAttribute(description = "Camel ID")
+    String getCamelId();
+
+    @ManagedAttribute(description = "Camel ManagementName")
+    String getCamelManagementName();
+
+    @ManagedAttribute(description = "Thread Pool ID")
+    String getId();
+
+    @ManagedAttribute(description = "ID of source for creating Thread Pool")
+    String getSourceId();
+
+    @ManagedAttribute(description = "Route ID for the source, which created the Thread Pool")
+    String getRouteId();
+
+    @ManagedAttribute(description = "ID of the thread pool profile which this pool is based upon")
+    String getThreadPoolProfileId();
+
+    @ManagedAttribute(description = "Whether this executor uses virtual threads")
+    boolean isVirtualThread();
+
+    @ManagedAttribute(description = "Is shutdown")
+    boolean isShutdown();
+}

core/camel-management/src/main/java/org/apache/camel/management/DefaultManagementObjectNameStrategy.java

@@ -17,6 +17,7 @@
 package org.apache.camel.management;
 
 import java.net.UnknownHostException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import javax.management.MalformedObjectNameException;
@@ -54,6 +55,7 @@
 import org.apache.camel.management.mbean.ManagedSupervisingRouteController;
 import org.apache.camel.management.mbean.ManagedThreadPool;
 import org.apache.camel.management.mbean.ManagedTracer;
+import org.apache.camel.management.mbean.ManagedVirtualThreadExecutor;
 import org.apache.camel.spi.DataFormat;
 import org.apache.camel.spi.EventNotifier;
 import org.apache.camel.spi.ManagementObjectNameStrategy;
@@ -162,6 +164,9 @@ public ObjectName getObjectName(Object managedObject) throws MalformedObjectName
             objectName = getObjectNameForTracer(mt.getContext(), mt.getTracer());
         } else if (managedObject instanceof ManagedThreadPool mes) {
             objectName = getObjectNameForThreadPool(mes.getContext(), mes.getThreadPool(), mes.getId(), mes.getSourceId());
+        } else if (managedObject instanceof ManagedVirtualThreadExecutor mvte) {
+            objectName = getObjectNameForThreadPool(
+                    mvte.getContext(), mvte.getExecutorService(), mvte.getId(), mvte.getSourceId());
         } else if (managedObject instanceof ManagedClusterService mcs) {
             objectName = getObjectNameForClusterService(mcs.getContext(), mcs.getService());
         } else if (managedObject instanceof ManagedService ms) {
@@ -422,6 +427,18 @@ public ObjectName getObjectNameForClusterService(CamelContext context, CamelClus
     public ObjectName getObjectNameForThreadPool(
             CamelContext context, ThreadPoolExecutor threadPool, String id, String sourceId)
             throws MalformedObjectNameException {
+        return getObjectNameForThreadPool(context, id, sourceId);
+    }
+
+    @Override
+    public ObjectName getObjectNameForThreadPool(
+            CamelContext context, ExecutorService executorService, String id, String sourceId)
+            throws MalformedObjectNameException {
+        return getObjectNameForThreadPool(context, id, sourceId);
+    }
+
+    private ObjectName getObjectNameForThreadPool(CamelContext context, String id, String sourceId)
+            throws MalformedObjectNameException {
         StringBuilder buffer = new StringBuilder();
         buffer.append(domainName).append(":");
         buffer.append(KEY_CONTEXT).append("=").append(getContextId(context)).append(",");

core/camel-management/src/main/java/org/apache/camel/management/DefaultManagementObjectStrategy.java

@@ -16,6 +16,7 @@
  */
 package org.apache.camel.management;
 
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import org.apache.camel.CamelContext;
@@ -111,6 +112,7 @@
 import org.apache.camel.management.mbean.ManagedTransformer;
 import org.apache.camel.management.mbean.ManagedUnmarshal;
 import org.apache.camel.management.mbean.ManagedValidate;
+import org.apache.camel.management.mbean.ManagedVirtualThreadExecutor;
 import org.apache.camel.management.mbean.ManagedWeightedLoadBalancer;
 import org.apache.camel.management.mbean.ManagedWireTapProcessor;
 import org.apache.camel.model.ExpressionNode;
@@ -284,6 +286,16 @@ public Object getManagedObjectForThreadPool(
         return mtp;
     }
 
+    @Override
+    public Object getManagedObjectForThreadPool(
+            CamelContext context, ExecutorService executorService,
+            String id, String sourceId, String routeId, String threadPoolProfileId) {
+        ManagedVirtualThreadExecutor mvte
+                = new ManagedVirtualThreadExecutor(context, executorService, id, sourceId, routeId, threadPoolProfileId);
+        mvte.init(context.getManagementStrategy());
+        return mvte;
+    }
+
     @Override
     public Object getManagedObjectForEventNotifier(CamelContext context, EventNotifier eventNotifier) {
         ManagedEventNotifier men = new ManagedEventNotifier(context, eventNotifier);

core/camel-management/src/main/java/org/apache/camel/management/JmxManagementLifecycleStrategy.java

@@ -24,6 +24,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import javax.management.JMException;
@@ -149,7 +150,7 @@ public class JmxManagementLifecycleStrategy extends ServiceSupport implements Li
     private final Set<String> knowRouteIds = new HashSet<>();
     private final Map<BacklogTracer, ManagedBacklogTracer> managedBacklogTracers = new HashMap<>();
     private final Map<DefaultBacklogDebugger, ManagedBacklogDebugger> managedBacklogDebuggers = new HashMap<>();
-    private final Map<ThreadPoolExecutor, Object> managedThreadPools = new HashMap<>();
+    private final Map<Object, Object> managedThreadPools = new HashMap<>();
 
     public JmxManagementLifecycleStrategy() {
     }
@@ -815,6 +816,62 @@ public void onThreadPoolRemove(CamelContext camelContext, ThreadPoolExecutor thr
         }
     }
 
+    @Override
+    public void onThreadPoolAdd(
+            CamelContext camelContext, ExecutorService executorService, String id,
+            String sourceId, String routeId, String threadPoolProfileId) {
+
+        if (!initialized) {
+            preServices
+                    .add(lf -> lf.onThreadPoolAdd(camelContext, executorService, id, sourceId, routeId,
+                            threadPoolProfileId));
+            return;
+        }
+
+        if (!shouldRegister(executorService, null)) {
+            return;
+        }
+
+        Object mtp = getManagementObjectStrategy().getManagedObjectForThreadPool(camelContext, executorService, id, sourceId,
+                routeId, threadPoolProfileId);
+        if (mtp == null) {
+            return;
+        }
+
+        if (getManagementStrategy().isManaged(mtp)) {
+            LOG.trace("The executor service is already managed: {}", executorService);
+            return;
+        }
+
+        try {
+            manageObject(mtp);
+            managedThreadPools.put(executorService, mtp);
+        } catch (Exception e) {
+            LOG.warn("Could not register executor service: {} as ThreadPool MBean.", executorService, e);
+        }
+    }
+
+    @Override
+    public void onThreadPoolRemove(CamelContext camelContext, ExecutorService executorService) {
+        if (!initialized) {
+            return;
+        }
+
+        Object mtp = managedThreadPools.remove(executorService);
+        if (mtp != null) {
+            if (!getManagementStrategy().isManaged(mtp)) {
+                LOG.trace("The executor service is not managed: {}", executorService);
+                return;
+            }
+
+            try {
+                unmanageObject(mtp);
+            } catch (Exception e) {
+                LOG.warn("Could not unregister ThreadPool MBean", e);
+            }
+        }
+    }
+
     @Override
     public void onRouteContextCreate(Route route) {
         // Create a map (ProcessorType -> PerformanceCounter)