camel.main.virtualThreadsEnabled=true

java -Dcamel.threads.virtual.enabled=true -jar myapp.jar

import org.eclipse.jetty.util.thread.VirtualThreadPool;

JettyHttpComponent jetty = context.getComponent("jetty", JettyHttpComponent.class);

// Create Jetty's VirtualThreadPool for request handling
VirtualThreadPool virtualThreadPool = new VirtualThreadPool();
virtualThreadPool.setName("CamelJettyVirtual");
jetty.setThreadPool(virtualThreadPool);

<bean id="jettyThreadPool" class="org.eclipse.jetty.util.thread.VirtualThreadPool">
    <property name="name" value="CamelJettyVirtual"/>
</bean>

<bean id="jetty" class="org.apache.camel.component.jetty.JettyHttpComponent">
    <property name="threadPool" ref="jettyThreadPool"/>
</bean>

from("platform-http:/api/orders")
    .threads()  // Offload to virtual thread pool
    .to("jpa:Order");  // Blocking JPA operation

<route>
    <from uri="platform-http:/api/orders"/>
    <threads/>
    <to uri="jpa:Order"/>
</route>

- route:
    from:
      uri: platform-http:/api/orders
      steps:
        - threads: {}
        - to:
            uri: jpa:Order

// With virtual threads, these blocking calls don't waste platform threads
from("seda:process?virtualThreadPerTask=true&concurrentConsumers=500")
    .to("jpa:Order")           // Blocking JDBC under the hood
    .to("sql:SELECT * FROM inventory WHERE id = :#${body.itemId}")
    .to("mongodb:orders");

<route>
    <from uri="seda:process?virtualThreadPerTask=true&amp;concurrentConsumers=500"/>
    <to uri="jpa:Order"/>
    <to uri="sql:SELECT * FROM inventory WHERE id = :#${body.itemId}"/>
    <to uri="mongodb:orders"/>
</route>

- route:
    from:
      uri: seda:process
      parameters:
        virtualThreadPerTask: true
        concurrentConsumers: 500
      steps:
        - to:
            uri: jpa:Order
        - to:
            uri: "sql:SELECT * FROM inventory WHERE id = :#${body.itemId}"
        - to:
            uri: mongodb:orders

from("seda:orders?concurrentConsumers=10")
    .process(this::processOrder)
    .to("direct:fulfillment");

<route>
    <from uri="seda:orders?concurrentConsumers=10"/>
    <process ref="processOrder"/>
    <to uri="direct:fulfillment"/>
</route>

- route:
    from:
      uri: seda:orders
      parameters:
        concurrentConsumers: 10
      steps:
        - process:
            ref: processOrder
        - to:
            uri: direct:fulfillment

from("seda:orders?virtualThreadPerTask=true&concurrentConsumers=100")
    .process(this::processOrder)  // I/O-bound operation
    .to("direct:fulfillment");

<route>
    <from uri="seda:orders?virtualThreadPerTask=true&amp;concurrentConsumers=100"/>
    <process ref="processOrder"/>
    <to uri="direct:fulfillment"/>
</route>

- route:
    from:
      uri: seda:orders
      parameters:
        virtualThreadPerTask: true
        concurrentConsumers: 100
      steps:
        - process:
            ref: processOrder
        - to:
            uri: direct:fulfillment

flowchart TB
    subgraph traditional["Traditional Model (Fixed Pool)"]
        direction TB
        Q1[("SEDA Queue")]
        C1["Consumer Thread 1"]
        C2["Consumer Thread 2"]
        C3["Consumer Thread N"]
        P1["Process Message"]

        Q1 -->|"poll()"| C1
        Q1 -->|"poll()"| C2
        Q1 -->|"poll()"| C3
        C1 --> P1
        C2 --> P1
        C3 --> P1
    end

    subgraph virtual["Virtual Thread Per Task Model"]
        direction TB
        Q2[("SEDA Queue")]
        COORD["Coordinator Thread"]
        SEM{{"Semaphore (concurrency limit)"}}
        VT1["Virtual Thread 1"]
        VT2["Virtual Thread 2"]
        VTN["Virtual Thread N"]
        P2["Process Message"]

        Q2 -->|"poll()"| COORD
        COORD -->|"acquire"| SEM
        SEM -->|"spawn"| VT1
        SEM -->|"spawn"| VT2
        SEM -->|"spawn"| VTN
        VT1 --> P2
        VT2 --> P2
        VTN --> P2
    end

camel.threads.virtual.enabled=true

java -Dcamel.threads.virtual.enabled=true -jar myapp.jar

// Queue holds up to 10,000 messages
from("seda:orders?size=10000")

<from uri="seda:orders?size=10000"/>

from:
  uri: seda:orders
  parameters:
    size: 10000

// Producer blocks up to 10 seconds when queue is full
from("direct:incoming")
    .to("seda:processing?size=5000&blockWhenFull=true&offerTimeout=10000");

<route>
    <from uri="direct:incoming"/>
    <to uri="seda:processing?size=5000&amp;blockWhenFull=true&amp;offerTimeout=10000"/>
</route>

- route:
    from:
      uri: direct:incoming
      steps:
        - to:
            uri: seda:processing
            parameters:
              size: 5000
              blockWhenFull: true
              offerTimeout: 10000

// Max 200 concurrent virtual threads processing messages
from("seda:orders?virtualThreadPerTask=true&concurrentConsumers=200")
    .to("http://downstream-service/api");

<route>
    <from uri="seda:orders?virtualThreadPerTask=true&amp;concurrentConsumers=200"/>
    <to uri="http://downstream-service/api"/>
</route>

- route:
    from:
      uri: seda:orders
      parameters:
        virtualThreadPerTask: true
        concurrentConsumers: 200
      steps:
        - to:
            uri: http://downstream-service/api

// Producer side: buffer up to 10,000, block if full (with timeout)
from("rest:post:/orders")
    .to("seda:order-queue?size=10000&blockWhenFull=true&offerTimeout=30000");

// Consumer side: process with virtual threads, max 500 concurrent
from("seda:order-queue?virtualThreadPerTask=true&concurrentConsumers=500")
    .to("http://inventory-service/check")
    .to("http://payment-service/process")
    .to("jpa:Order");

<!-- Producer side: buffer up to 10,000, block if full (with timeout) -->
<route>
    <from uri="rest:post:/orders"/>
    <to uri="seda:order-queue?size=10000&amp;blockWhenFull=true&amp;offerTimeout=30000"/>
</route>

<!-- Consumer side: process with virtual threads, max 500 concurrent -->
<route>
    <from uri="seda:order-queue?virtualThreadPerTask=true&amp;concurrentConsumers=500"/>
    <to uri="http://inventory-service/check"/>
    <to uri="http://payment-service/process"/>
    <to uri="jpa:Order"/>
</route>

# Producer side: buffer up to 10,000, block if full (with timeout)
- route:
    from:
      uri: rest:post:/orders
      steps:
        - to:
            uri: seda:order-queue
            parameters:
              size: 10000
              blockWhenFull: true
              offerTimeout: 30000

# Consumer side: process with virtual threads, max 500 concurrent
- route:
    from:
      uri: seda:order-queue
      parameters:
        virtualThreadPerTask: true
        concurrentConsumers: 500
      steps:
        - to:
            uri: http://inventory-service/check
        - to:
            uri: http://payment-service/process
        - to:
            uri: jpa:Order

public class OrderProcessingRoute extends RouteBuilder {
    @Override
    public void configure() {
        // Receive orders via REST, queue them for async processing
        // Block callers if queue is full (with 30s timeout)
        rest("/orders")
            .post()
            .to("seda:incoming-orders?size=10000&blockWhenFull=true&offerTimeout=30000");

        // Process with virtual threads - each order gets its own thread
        // Limit to 500 concurrent to protect downstream services
        from("seda:incoming-orders?virtualThreadPerTask=true&concurrentConsumers=500")
            .routeId("order-processor")
            .log("Processing order ${body.orderId} on ${threadName}")
            .to("http://inventory-service/check")      // I/O - virtual thread parks
            .to("http://payment-service/process")      // I/O - virtual thread parks
            .to("jpa:Order")                           // I/O - virtual thread parks
            .to("direct:send-confirmation");
    }
}

# Platform threads, fixed pool
mvn test -Dtest=VirtualThreadsLoadTest -pl core/camel-core

# Virtual threads, fixed pool
mvn test -Dtest=VirtualThreadsLoadTest -pl core/camel-core \
    -Dcamel.threads.virtual.enabled=true

# Virtual threads, thread-per-task (optimal)
mvn test -Dtest=VirtualThreadsLoadTest -pl core/camel-core \
    -Dcamel.threads.virtual.enabled=true \
    -Dloadtest.virtualThreadPerTask=true

import org.apache.camel.util.concurrent.ContextValue;

// Create a context value (picks ScopedValue or ThreadLocal automatically)
private static final ContextValue<String> TENANT_ID = ContextValue.newInstance("tenantId");

// Bind a value for a scope
ContextValue.where(TENANT_ID, "acme-corp", () -> {
    // Code here can access TENANT_ID.get()
    processRequest();
    return result;
});

// Inside processRequest(), on any thread in the scope:
public void processRequest() {
    String tenant = TENANT_ID.get();  // Returns "acme-corp"
    // ... process with tenant context
}

// Use ContextValue.newInstance() for READ-ONLY context passing
private static final ContextValue<RequestContext> REQUEST_CTX = ContextValue.newInstance("requestCtx");

// Use ContextValue.newThreadLocal() when you need MUTABLE state
private static final ContextValue<Counter> COUNTER = ContextValue.newThreadLocal("counter", Counter::new);

// Example: Passing context during route creation
private static final ContextValue<ProcessorDefinition<?>> CREATE_PROCESSOR
    = ContextValue.newInstance("CreateProcessor");

// When creating processors, bind the context
ContextValue.where(CREATE_PROCESSOR, this, () -> {
    return createOutputsProcessor(routeContext);
});

// Child code can access the current processor being created
ProcessorDefinition<?> current = CREATE_PROCESSOR.orElse(null);

// Before: ThreadLocal
private static final ThreadLocal<User> CURRENT_USER = new ThreadLocal<>();

public void handleRequest(User user) {
    CURRENT_USER.set(user);
    try {
        processRequest();
    } finally {
        CURRENT_USER.remove();
    }
}

// After: ContextValue
private static final ContextValue<User> CURRENT_USER = ContextValue.newInstance("currentUser");

public void handleRequest(User user) {
    ContextValue.where(CURRENT_USER, user, this::processRequest);
}

# Start with virtual threads disabled, benchmark, then enable
camel.threads.virtual.enabled=false

# When enabling, test thoroughly
camel.threads.virtual.enabled=true

// For I/O-bound: use virtualThreadPerTask with high concurrency limit
from("seda:io-bound?virtualThreadPerTask=true&concurrentConsumers=1000")

// For CPU-bound: stick with traditional model, tune pool size
from("seda:cpu-bound?concurrentConsumers=4")  // ~number of CPU cores

<!-- For I/O-bound: use virtualThreadPerTask with high concurrency limit -->
<from uri="seda:io-bound?virtualThreadPerTask=true&amp;concurrentConsumers=1000"/>

<!-- For CPU-bound: stick with traditional model, tune pool size -->
<from uri="seda:cpu-bound?concurrentConsumers=4"/>

# For I/O-bound: use virtualThreadPerTask with high concurrency limit
from:
  uri: seda:io-bound
  parameters:
    virtualThreadPerTask: true
    concurrentConsumers: 1000

# For CPU-bound: stick with traditional model, tune pool size
from:
  uri: seda:cpu-bound
  parameters:
    concurrentConsumers: 4

// Avoid: can pin virtual thread
synchronized (lock) {
    doBlockingOperation();
}

// Prefer: virtual thread can unmount
lock.lock();
try {
    doBlockingOperation();
} finally {
    lock.unlock();
}

VirtualThread[#123]/Camel (camel-1) thread #5 - seda://orders

# Record virtual thread events
java -XX:StartFlightRecording=filename=recording.jfr,settings=default \
     -Dcamel.threads.virtual.enabled=true \
     -jar myapp.jar

# Log when virtual threads pin (JDK 21+)
java -Djdk.tracePinnedThreads=short \
     -Dcamel.threads.virtual.enabled=true \
     -jar myapp.jar

public class RestApiRoute extends RouteBuilder {
    @Override
    public void configure() {
        // REST endpoint receives requests
        rest("/api")
            .post("/orders")
            .to("seda:process-order");

        // Process with virtual threads - handle 1000s of concurrent requests
        from("seda:process-order?virtualThreadPerTask=true&concurrentConsumers=2000")
            .routeId("order-processor")
            // Each step may block on I/O - virtual threads park efficiently
            .to("http://inventory-service/reserve")
            .to("http://payment-service/charge")
            .to("jpa:Order?persistenceUnit=orders")
            .to("kafka:order-events");
    }
}

public class ParallelEnrichmentRoute extends RouteBuilder {
    @Override
    public void configure() {
        from("direct:enrich")
            .multicast()
                .parallelProcessing()
                .executorService(virtualThreadExecutor())  // Use virtual threads
                .to("direct:enrichFromUserService",
                    "direct:enrichFromOrderHistory",
                    "direct:enrichFromRecommendations")
            .end()
            .to("direct:aggregate");
    }

    private ExecutorService virtualThreadExecutor() {
        return getCamelContext()
            .getExecutorServiceManager()
            .newCachedThreadPool(this, "enrichment");
        // When camel.threads.virtual.enabled=true, this returns a virtual thread executor
    }
}

public class TenantAwareRoute extends RouteBuilder {

    private static final ContextValue<String> TENANT_ID = ContextValue.newInstance("tenantId");

    @Override
    public void configure() {
        // ContextValue is scoped to the current thread - it works within a single
        // route or call chain, not across asynchronous boundaries like SEDA queues.
        // For cross-route context, use exchange properties instead.
        from("platform-http:/api/{tenant}/orders")
            .process(exchange -> {
                String tenant = exchange.getMessage().getHeader("tenant", String.class);
                exchange.setProperty("tenantId", tenant);
            })
            .to("seda:process");

        from("seda:process?virtualThreadPerTask=true&concurrentConsumers=500")
            .process(exchange -> {
                // Use exchange properties for context that crosses async boundaries
                String tenant = exchange.getProperty("tenantId", "default", String.class);
                log.info("Processing for tenant: {}", tenant);
            })
            .toD("jpa:Order?persistenceUnit=${exchangeProperty.tenantId}");
    }
}