CAMEL-20199: Add JMX support for virtual thread executors

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/seda.json

@@ -38,7 +38,7 @@
   "properties": {
     "name": { "index": 0, "kind": "path", "displayName": "Name", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Name of queue" },
     "size": { "index": 1, "kind": "parameter", "displayName": "Size", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000, "description": "The maximum capacity of the SEDA queue (i.e., the number of messages it can hold). Will by default use the queueSize set on the SEDA component." },
-    "concurrentConsumers": { "index": 2, "kind": "parameter", "displayName": "Concurrent Consumers", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "description": "Number of concurrent threads processing exchanges." },
+    "concurrentConsumers": { "index": 2, "kind": "parameter", "displayName": "Concurrent Consumers", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "description": "Number of concurrent threads processing exchanges. When virtualThreadPerTask is enabled, this becomes a concurrency limit (0 = unlimited) and defaults to 0 instead of 1." },
     "bridgeErrorHandler": { "index": 3, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." },
     "exceptionHandler": { "index": 4, "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored." },
     "exchangePattern": { "index": 5, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." },

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/stub.json

@@ -40,7 +40,7 @@
   "properties": {
     "name": { "index": 0, "kind": "path", "displayName": "Name", "group": "common", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Name of queue" },
     "size": { "index": 1, "kind": "parameter", "displayName": "Size", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000, "description": "The maximum capacity of the SEDA queue (i.e., the number of messages it can hold). Will by default use the queueSize set on the SEDA component." },
-    "concurrentConsumers": { "index": 2, "kind": "parameter", "displayName": "Concurrent Consumers", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "description": "Number of concurrent threads processing exchanges." },
+    "concurrentConsumers": { "index": 2, "kind": "parameter", "displayName": "Concurrent Consumers", "group": "consumer", "label": "consumer", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "description": "Number of concurrent threads processing exchanges. When virtualThreadPerTask is enabled, this becomes a concurrency limit (0 = unlimited) and defaults to 0 instead of 1." },
     "bridgeErrorHandler": { "index": 3, "kind": "parameter", "displayName": "Bridge Error Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions (if possible) occurred while the Camel consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. Important: This is only possible if the 3rd party component allows Camel to be alerted if an exception was thrown. Some components handle this internally only, and therefore bridgeErrorHandler is not possible. In other situations we may improve the Camel component to hook into the 3rd party component and make this possible for future releases. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored." },
     "exceptionHandler": { "index": 4, "kind": "parameter", "displayName": "Exception Handler", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "object", "javaType": "org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.", "deprecated": false, "autowired": false, "secret": false, "description": "To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this option is not in use. By default the consumer will deal with exceptions, that will be logged at WARN or ERROR level and ignored." },
     "exchangePattern": { "index": 5, "kind": "parameter", "displayName": "Exchange Pattern", "group": "consumer (advanced)", "label": "consumer,advanced", "required": false, "type": "enum", "javaType": "org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ], "deprecated": false, "autowired": false, "secret": false, "description": "Sets the exchange pattern when the consumer creates an exchange." },

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);