Apache Camel

---
name: apache-camel
description: Building or debugging Camel routes, EIP integrations, or Camel on Spring Boot
---

# Apache Camel Advanced Skill — v1.1

## Quick Card

> Read this card first. Load a section below only when the task needs it.

| | |
|---|---|
| **Use when** | Building or debugging Camel routes, EIP integrations, or Camel on Spring Boot |
| **Skip when** | Plain request/response HTTP calls with no routing — a client library is simpler |
| **Inputs** | Source/target endpoints, message formats, error and retry policy |
| **Produces** | `RouteBuilder` classes, processors, route tests |
| **Steps** | 1. Model Exchange → Message → Body/Headers → 2. Write the route in Java DSL → 3. Add `onException` per failure class → 4. Extract logic into processors → 5. Test with `AdviceWith` / mock endpoints |
| **Done when** | Route tests pass with endpoints mocked; every exception class has an explicit handler |
| **Senior defaults** | Retryable vs non-retryable errors get separate `onException` blocks · exponential redelivery with `handled(true)` + dead-letter route · no business logic inline in the DSL — processors · headers for routing, body for payload · idempotent consumer on at-least-once sources |
| **Load on demand** | §1 mental model · §2 route DSL · §3 error handling · §4 processors · §5 Spring Boot · §6 testing · §7 debugging |
| **Run report** | `html_report_skill` — adds: Route map · Error-handling matrix |
| **Pairs with** | `error_handling_skill`, `spring_advanced_skill`, `logger_skill` |

---

## 1. Core Concepts (Mental Model First)

Understanding Camel starts with four objects that flow through every route:

```java
Exchange  — the wrapper that carries a message through a route
  └── Message (IN)   — the current message being processed
        ├── Body     — the payload (String, byte[], POJO, InputStream, etc.)
        ├── Headers  — key/value metadata (like HTTP headers, file names, etc.)
        └── Properties — exchange-scoped metadata (survive routing steps)
```

**Route** = a pipeline of steps. Data enters at a `from()`, flows through
processors and transformers, and exits at one or more `to()` endpoints.

**Component** = a connector to an external system (file, HTTP, Kafka, DB, etc.).
A component creates **Endpoints**. An endpoint creates **Producers** (send) and
**Consumers** (receive).

**Processor** = a single processing step. Receives an `Exchange`, modifies it,
passes it on.

---

## 2. Route DSL — Java (Primary)

### Basic Route Structure

```java
import org.apache.camel.builder.RouteBuilder;

/**
 * Processes incoming orders from a file drop folder.
 *
 * <p>Reads JSON files from /data/orders/incoming, validates them,
 * transforms to the internal format, and sends to the order-processing queue.
 * Failed files are moved to /data/orders/error.</p>
 */
@Component
public class OrderIngestionRoute extends RouteBuilder {

    @Override
    public void configure() throws Exception {

        // ── Global error handling for this route builder ──────────────
        // If anything throws an exception, move the file to the error folder
        // and send an alert. Do NOT retry file routes (files don't disappear).
        errorHandler(
            deadLetterChannel("file:/data/orders/error?autoCreate=true")
                .log("Order file failed: ${exception.message}")
        );

        // ── Route definition ──────────────────────────────────────────
        from("file:/data/orders/incoming?delete=true&moveFailed=../error")
            .routeId("order-ingestion")           // give every route a unique ID
            .description("Reads order files and sends to processing queue")

            .log("Processing order file: ${header.CamelFileName}")

            // Step 1: Unmarshal JSON to Java object
            .unmarshal().json(OrderFileDto.class)

            // Step 2: Validate (custom Processor — see section 4)
            .process(new OrderValidator())

            // Step 3: Transform to internal domain object
            .bean(OrderTransformer.class, "transform")

            // Step 4: Send to queue for async processing
            .to("pulsar:persistent://orders/default/order-processing")

            .log("Order ${body.orderId} sent to processing queue");
    }
}
```

### Content-Based Router (EIP Pattern)

```java
/**
 * Routes payments to the correct processor based on payment type.
 *
 * <p>Uses Camel's Content-Based Router pattern — one of the core EIPs.
 * Each payment type has its own downstream endpoint so they can be
 * scaled and managed independently.</p>
 */
@Component
public class PaymentRoutingRoute extends RouteBuilder {

    @Override
    public void configure() {

        from("direct:route-payment")
            .routeId("payment-routing")
            .description("Routes payments to the correct processor by type")

            .choice()
                // Card payments → card gateway
                .when(header("paymentType").isEqualTo("CARD"))
                    .to("direct:process-card-payment")

                // Bank transfers → BACS processor
                .when(header("paymentType").isEqualTo("BANK_TRANSFER"))
                    .to("direct:process-bank-transfer")

                // PayPal → PayPal adapter
                .when(header("paymentType").isEqualTo("PAYPAL"))
                    .to("direct:process-paypal")

                // Unknown type — send to dead letter for manual review
                .otherwise()
                    .log(LoggingLevel.WARN, "Unknown payment type: ${header.paymentType}")
                    .to("direct:unknown-payment-type")
            .end();
    }
}
```

### Splitter + Aggregator (Process a list, collect results)

```java
/**
 * Splits a batch order into individual items, processes each one,
 * then aggregates the results back into a single batch result.
 */
@Component
public class BatchOrderRoute extends RouteBuilder {

    @Override
    public void configure() {

        from("direct:process-batch-order")
            .routeId("batch-order-processing")

            // Split the order into individual line items
            // ${body.items} is a List<OrderItem>
            .split(simple("${body.items}"), new BatchResultAggregator())
                .parallelProcessing()         // process items in parallel
                .streaming()                  // don't load all into memory at once

                // Process each individual item
                .process(exchange -> {
                    OrderItem item = exchange.getIn().getBody(OrderItem.class);
                    // ... processing logic
                })

                .to("direct:process-single-item")

            .end()

            // After all items processed, body is the aggregated BatchResult
            .log("Batch complete: ${body.successCount} succeeded, ${body.failureCount} failed");
    }
}
```

---

## 3. Error Handling — The Right Way

### Per-Route Exception Handling

```java
@Component
public class PaymentProcessingRoute extends RouteBuilder {

    @Override
    public void configure() {

        // ── Handle specific exceptions ────────────────────────────────

        // Retryable error: network blip — retry up to 3 times with backoff
        onException(HttpOperationFailedException.class)
            .maximumRedeliveries(3)
            .redeliveryDelay(1000)          // 1 second initial delay
            .backOffMultiplier(2.0)         // double delay each retry: 1s, 2s, 4s
            .retryAttemptedLogLevel(LoggingLevel.WARN)
            .handled(true)                  // don't propagate after retries exhausted
            .to("direct:payment-failed");   // send to failure handler

        // Non-retryable error: validation failure — don't retry, just route to error
        onException(PaymentValidationException.class)
            .handled(true)
            .log(LoggingLevel.ERROR, "Payment validation failed: ${exception.message}")
            .setHeader("failureReason", simple("${exception.message}"))
            .to("direct:invalid-payment");

        // Catch-all: unexpected errors — log full stack trace and alert
        onException(Exception.class)
            .handled(true)
            .log(LoggingLevel.ERROR, "Unexpected error in payment route: ${exception.stacktrace}")
            .to("direct:payment-system-error");

        // ── Main route ────────────────────────────────────────────────
        from("direct:process-payment")
            .routeId("payment-processing")
            .process(new PaymentValidator())
            .to("https://payment-gateway/charge")
            .to("direct:payment-success");
    }
}
```

### Dead Letter Channel (Global Fallback)

```java
// In a base RouteBuilder that other routes extend:
errorHandler(
    deadLetterChannel("jms:queue:dead-letter")
        .maximumRedeliveries(3)
        .redeliveryDelay(2000)
        .useOriginalMessage()           // send the original message, not the modified one
        .deadLetterHandleNewException(false)
        .log("Message sent to DLQ after ${header.CamelRedeliveryCounter} retries")
);
```

---

## 4. Processors — Custom Logic Steps

```java
/**
 * Validates that an incoming order has the required fields.
 *
 * <p>Implements {@link Processor} — the basic unit of work in a Camel route.
 * Throws {@link OrderValidationException} if validation fails, which triggers
 * the route's error handling.</p>
 */
public class OrderValidator implements Processor {

    private static final Logger log = LoggerFactory.getLogger(OrderValidator.class);

    @Override
    public void process(Exchange exchange) throws Exception {
        // Get the current message body, converted to our domain type
        OrderRequest order = exchange.getIn().getBody(OrderRequest.class);

        log.debug("Validating order from customer {}", order.getCustomerId());

        // Validate — throw a domain exception if invalid (triggers error handling)
        if (order.getCustomerId() == null || order.getCustomerId() <= 0) {
            throw new OrderValidationException("Customer ID is required and must be positive");
        }
        if (order.getItems() == null || order.getItems().isEmpty()) {
            throw new OrderValidationException("Order must contain at least one item");
        }

        // Enrich the exchange with metadata for downstream steps
        exchange.getIn().setHeader("validatedAt", System.currentTimeMillis());
        exchange.getIn().setHeader("customerId", order.getCustomerId());

        log.debug("Order validation passed for customer {}", order.getCustomerId());
    }
}
```

---

## 5. Spring Boot Integration

### application.yml for Camel

```yaml
camel:
  springboot:
    name: OrderService              # shown in JMX, logs, and health endpoint
    main-run-controller: true       # keeps the app alive for consumer routes
  health:
    enabled: true                   # exposes /actuator/health with route statuses
  management:
    enabled: true                   # exposes Camel routes via JMX

# Component-level config (set per component)
camel:
  component:
    http:
      connection-timeout: 5000      # ms — ALWAYS set timeouts on HTTP components
      response-timeout: 10000
```

### Inject Camel Context in Spring Beans

```java
@Service
@RequiredArgsConstructor
public class OrderDispatchService {

    private final ProducerTemplate producerTemplate;  // injected by Spring (Camel auto-configures it)

    /**
     * Sends an order synchronously to the order-processing route.
     *
     * @param order the order to process
     * @return the processed order result
     * @throws CamelExecutionException if the route throws an exception
     */
    public OrderResult dispatchOrder(OrderRequest order) {
        // Send to a direct: endpoint and wait for the result
        return producerTemplate.requestBody("direct:process-order", order, OrderResult.class);
    }

    /**
     * Sends an order asynchronously — fire and forget.
     *
     * @param order the order to process
     */
    public void dispatchOrderAsync(OrderRequest order) {
        producerTemplate.asyncSendBody("direct:process-order", order);
    }
}
```

---

## 6. Testing Camel Routes

```java
/**
 * Unit tests for OrderIngestionRoute.
 *
 * <p>Uses CamelTestSupport — starts a lightweight Camel context with
 * the route under test. Mock endpoints replace real external systems
 * so tests run fast without network or filesystem access.</p>
 */
@CamelSpringBootTest
@SpringBootTest
class OrderIngestionRouteTest extends CamelTestSupport {

    @Autowired
    private ProducerTemplate producerTemplate;

    @EndpointInject("mock:pulsar:persistent://orders/default/order-processing")
    private MockEndpoint mockOrderQueue;

    @Test
    @DisplayName("Given valid order file, when route processes it, then sends to queue")
    void givenValidOrderFile_whenRouteProcesses_thenSendsToQueue() throws Exception {
        // Arrange — set expectation on the mock endpoint
        mockOrderQueue.expectedMessageCount(1);
        mockOrderQueue.expectedHeaderReceived("customerId", 42L);

        String orderJson = """
            {
                "customerId": 42,
                "items": [{"productId": 1, "quantity": 2}]
            }
            """;

        // Act — send directly to the route's from() endpoint
        producerTemplate.sendBodyAndHeader(
            "file:/data/orders/incoming",
            orderJson,
            Exchange.FILE_NAME,
            "order-123.json"
        );

        // Assert — verify the mock received exactly what we expected
        mockOrderQueue.assertIsSatisfied();
    }

    @Test
    @DisplayName("Given invalid order, when route processes it, then sends to error endpoint")
    void givenInvalidOrder_whenRouteProcesses_thenSendsToError() throws Exception {
        MockEndpoint mockError = getMockEndpoint("mock:direct:invalid-payment");
        mockError.expectedMessageCount(1);

        producerTemplate.sendBody("direct:process-order", "{ }");  // empty JSON = invalid

        mockError.assertIsSatisfied();
    }

    @Override
    protected RoutesBuilder createRouteBuilder() {
        return new OrderIngestionRoute();
    }
}
```

---

## 7. Debugging Guide

### Logging the Exchange at Any Point

```java
// Add .log() anywhere in a route to inspect the message
from("direct:my-route")
    .log("Body before transform: ${body}")
    .log("Headers: ${headers}")
    .bean(MyTransformer.class)
    .log("Body after transform: ${body}")
    .to("direct:next");
```

### Tracer (Full Exchange History)

```java
// In a @Configuration class — enables detailed tracing in dev
@Bean
public BacklogTracer backlogTracer(CamelContext context) {
    BacklogTracer tracer = BacklogTracer.createTracer(context);
    tracer.setEnabled(true);
    tracer.setBacklogSize(100);          // keep last 100 traced exchanges
    tracer.setTraceFilter("routeId == 'order-ingestion'");  // trace specific route
    return tracer;
}
```

### Common Debugging Patterns

```java
// Pattern 1: Inspect body type at runtime
.process(exchange -> {
    Object body = exchange.getIn().getBody();
    System.out.println("Body type: " + body.getClass().getName());
    System.out.println("Body value: " + body);
})

// Pattern 2: Save body to a header before transforming (for error recovery)
.setHeader("originalBody", body())
.bean(RiskyTransformer.class)

// Pattern 3: Check exchange for errors
.process(exchange -> {
    Exception ex = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class);
    if (ex != null) {
        // handle the caught exception
    }
})
```

### Common Issues and Fixes

| Problem | Symptom | Fix |
|---------|---------|-----|
| Body is `null` after a `.to()` | Next processor gets null body | InOut vs InOnly MEP mismatch — check `.setExchangePattern(ExchangePattern.InOut)` |
| Route never starts | No log output | Check `autoStartup`, route ID conflicts, or missing `@Component` |
| File route re-processes files | Same file processed multiple times | Use `delete=true` or `move=../done` on the file component URI |
| `TypeConversionException` | Can't convert body to target type | Register a custom `TypeConverter` or add a `.convertBodyTo(TargetClass.class)` step |
| Memory grows on Splitter | OOM on large files | Add `.streaming()` to the split — loads one item at a time |
| `HTTP 503` not retried | Camel doesn't retry | Add `onException(HttpOperationFailedException.class).maximumRedeliveries(3)` |