Apache Camel で Kafka 送信失敗をリトライする（例外検知 + メタ情報チェック付き）【Camel 4 / XML DSL】

## はじめに

Apache Camel を使って Kafka にメッセージを送信する際、  
**例外や送信結果（RecordMetadata）の null などに対して適切にリトライ処理を実装する**ことは、信頼性の高いシステム構築に欠かせません。

本記事では、以下のような構成を実現します：

### やりたいこと

- Kafka 送信時に `KafkaException` などの例外が発生したらキャッチしてリトライ
- Kafka 送信成功時でも `RecordMetadata` が `null` または空の場合はリトライ
- 例外は自前の `RetryableKafkaException` を使って明確に制御
- XML DSL + Spring Boot + Apache Camel で構成

---

## 使用技術

- **Apache Camel 4.x**
- **Spring Boot 3.x**
- Kafka
- XML DSL

---

## ディレクトリ構成（概要）
src/
├─ main/
│  ├─ java/com/example/
│  │  ├─ KafkaSendApplication.java
│  │  ├─ exception/RetryableKafkaException.java
│  │  └─ processor/KafkaToRetryableExceptionMapper.java
│  ├─ resources/
│  │  ├─ application.properties
│  │  └─ camel-context.xml
pom.xml

---

## 第1章：application.properties（基本設定）

```properties
camel.component.kafka.brokers=localhost:9092
retry.max=3
retry.delay.ms=2000
```

## 第2章：pom.xml（Maven依存関係）

Apache Camel と Kafka を連携させるには、以下の依存関係を `pom.xml` に追加します。

```xml
<dependencies>
    <!-- Apache Camel のコアライブラリ -->
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-core</artifactId>
        <version>4.8.0</version>
    </dependency>

    <!-- Kafka コンポーネント -->
    <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-kafka</artifactId>
        <version>4.8.0</version>
    </dependency>

    <!-- Spring Boot との連携用 -->
    <dependency>
        <groupId>org.apache.camel.springboot</groupId>
        <artifactId>camel-spring-boot-starter</artifactId>
        <version>4.8.0</version>
    </dependency>
</dependencies>
```
## 第3章：RetryableKafkaException（自前例外クラス）

Kafka送信時の失敗や、送信後に `RecordMetadata` が存在しない（＝隠れた失敗）ケースを検出した際に、  
明示的にリトライ対象として扱うための **自前の例外クラス** を作成します。

このクラスを使って例外の意味を明確にし、Camel の `onException` によるリトライ制御を容易にします。

### `RetryableKafkaException.java`

```java
package com.example.exception;

/**
 * Kafka送信失敗時に再送を試みるための明示的な例外クラス。
 */
public class RetryableKafkaException extends RuntimeException {

    public RetryableKafkaException(String message) {
        super(message);
    }

    public RetryableKafkaException(String message, Throwable cause) {
        super(message, cause);
    }
}
```
## 第4章：KafkaToRetryableExceptionMapper（例外変換プロセッサ）

Kafkaへのメッセージ送信時に `KafkaException` が発生した場合、  
それをそのまま扱うのではなく、**自前の `RetryableKafkaException` にラップして再送出**することで、  
後段の `onException` でリトライ制御を統一的に扱えるようにします。

そのためのプロセッサがこの `KafkaToRetryableExceptionMapper` です。

### `KafkaToRetryableExceptionMapper.java`

```java
package com.example.processor;

import com.example.exception.RetryableKafkaException;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;

/**
 * Kafka送信時のKafkaExceptionを、RetryableKafkaExceptionに変換するProcessor。
 */
public class KafkaToRetryableExceptionMapper implements Processor {

    @Override
    public void process(Exchange exchange) {
        // Kafka送信時に発生した例外を取得
        Exception cause = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class);

        // 自前のリトライ対象例外に変換して送出
        throw new RetryableKafkaException("Kafka send failed", cause);
    }
}
```
## 第5章：camel-context.xml（ルート定義）

Kafka送信処理の中心となる Camel のルート定義を XML DSL で記述します。  
ここでは以下のポイントを重視しています：

- Kafka送信中に `KafkaException` が発生したら自前のリトライ可例外に変換
- Kafka送信成功後でも `RecordMetadata` が null または空なら明示的に失敗扱い
- それらの失敗はすべて `RetryableKafkaException` として統一的に扱い、再送処理へ

### `camel-context.xml`

```xml
<routes xmlns="http://camel.apache.org/schema/spring">

    <!-- RetryableKafkaException をリトライ対象に定義（最大3回、2秒間隔） -->
    <onException>
        <exception>com.example.exception.RetryableKafkaException</exception>
        <redeliveryPolicy maximumRedeliveries="{{retry.max}}" redeliveryDelay="{{retry.delay.ms}}" />
        <log message="RetryableKafkaException caught: ${exception.message}" loggingLevel="WARN" />
        <handled>
            <constant>true</constant>
        </handled>
    </onException>

    <!-- Kafka送信ルート -->
    <route id="kafka-send-route">
        <from uri="direct:kafkaSend" />
        <log message="Sending to Kafka: ${body}" />

        <!-- Kafka送信中にKafkaExceptionが発生したら、自前例外に変換 -->
        <onException>
            <exception>org.apache.camel.component.kafka.KafkaException</exception>
            <handled><constant>true</constant></handled>
            <process>
                <bean class="com.example.processor.KafkaToRetryableExceptionMapper"/>
            </process>
        </onException>

        <!-- Kafka送信 -->
        <to uri="kafka:my-topic" />

        <!-- Kafkaの送信結果を検証：RecordMetadata が無いなら失敗扱い -->
        <choice>
            <when>
                <or>
                    <simple>${header.CamelKafkaRecordMeta} == null</simple>
                    <simple>${header.CamelKafkaRecordMeta} == ''</simple>
                </or>
                <throwException exceptionType="com.example.exception.RetryableKafkaException">
                    <message>Kafka RecordMetadata is missing — send failed</message>
                </throwException>
            </when>
            <otherwise>
                <log message="Kafka send succeeded. Metadata: ${header.CamelKafkaRecordMeta}" />
            </otherwise>
        </choice>
    </route>

    <!-- 起動時に1回だけメッセージを送信するルート -->
    <route id="test-trigger">
        <from uri="timer:test?repeatCount=1" />
        <setBody>
            <constant>{"event":"test"}</constant>
        </setBody>
        <to uri="direct:kafkaSend" />
    </route>

</routes>
```
## 第6章：KafkaSendApplication（起動クラス）

このアプリケーションは Spring Boot ベースで構成されており、  
`main()` メソッドから Camel のルートが自動起動されます。

### `KafkaSendApplication.java`

```java
package com.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * Kafka送信アプリケーションのエントリーポイント。
 * Camel のルート（camel-context.xml）が自動で読み込まれます。
 */
@SpringBootApplication
public class KafkaSendApplication {
    public static void main(String[] args) {
        SpringApplication.run(KafkaSendApplication.class, args);
    }
}
```

## 第7章：動作結果（ログ）

この構成でアプリケーションを起動すると、Camel ルートが1回だけ実行され、Kafka にテストメッセージが送信されます。

正常に送信された場合、以下のようなログが出力されます：

Sending to Kafka: {“event”:“test”}
Kafka send succeeded. Metadata: KafkaProducerRecordMetadata(topic=‘my-topic’, partition=0, offset=123)

### 送信失敗時（例：Kafka ブローカーが起動していない場合）

Sending to Kafka: {“event”:“test”}
RetryableKafkaException caught: Kafka send failed
RetryableKafkaException caught: Kafka send failed
RetryableKafkaException caught: Kafka send failed

> 上記のように `onException` によって最大3回まで自動でリトライされます（`retry.max=3` の設定）。

### メタ情報なし（RecordMetadata が null or 空）の場合

Sending to Kafka: {“event”:“test”}
RetryableKafkaException caught: Kafka RecordMetadata is missing — send failed

このように Camel のルート上で Kafka 送信処理を安全にハンドリングできていることが確認できます。