WSO2 Kafka入站端点错误重试：外部系统恢复后消息重发咨询

你遇到的核心问题是：外部系统宕机时，Kafka入站端点拉取的消息转发失败，而外部系统恢复后这些失败消息没有自动重新投递。这主要是因为当前配置里自动提交了Kafka Offset，且缺少重试和失败兜底机制。下面结合你的现有配置，给出具体的调整方案：

1. 禁用自动Offset提交，改为手动提交 #

默认配置里的dual.commit.enabled=true会导致即使消息转发失败，Kafka Offset也被提交，后续不会再重新推送这些消息。我们需要修改入站端点参数，让Offset只有在消息成功发送后才提交：

修改后的入站端点配置：

<?xml version="1.0" encoding="UTF-8"?>
<inboundEndpoint name="EventTransmitter" protocol="kafka" sequence="transmit_sequence" suspend="false" onError="error_handling_sequence" xmlns="http://ws.apache.org/ns/synapse">
<parameters>
<parameter name="interval">10</parameter>
<parameter name="coordination">true</parameter>
<parameter name="sequential">true</parameter>
<parameter name="zookeeper.connect">localhost:2181</parameter>
<parameter name="consumer.type">highlevel</parameter>
<parameter name="content.type">application/json</parameter>
<parameter name="topics">event_topic</parameter>
<parameter name="group.id">myconsumer</parameter>
<parameter name="consumer.id">myconsumer</parameter>
<parameter name="dual.commit.enabled">false</parameter> <!-- 禁用双提交 -->
<parameter name="auto.commit.enable">false</parameter> <!-- 禁用自动提交 -->
<parameter name="auto.offset.reset">smallest</parameter> <!-- 重启后拉取未提交的消息 -->
</parameters>
</inboundEndpoint>

2. 在转发序列中添加重试机制 #

给transmit_sequence加上retry中介器，配置重试次数和间隔，应对外部系统临时不可用的场景：

修改后的转发序列：

<?xml version="1.0" encoding="UTF-8"?>
<sequence name="transmit_sequence" onError="error_handling_sequence" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
    <!-- 最多重试3次，每次间隔2秒 -->
    <retry count="3" interval="2000">
        <onRetryFailure>
            <log level="full">
                <property name="RETRY_INFO" value="Retrying message delivery, attempt in progress..."/>
            </log>
        </onRetryFailure>
        <send receive="event_transmit_out_sequence">
            <endpoint key="gov:endpoints/HandlerEndpoint.xml"/>
        </send>
    </retry>
</sequence>

3. 配置错误处理序列与死信队列 #

当重试多次仍失败时，将消息路由到死信队列（DLQ），避免阻塞正常消息流，同时保留失败消息以便后续处理：

创建错误处理序列error_handling_sequence：

<?xml version="1.0" encoding="UTF-8"?>
<sequence name="error_handling_sequence" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
    <log level="full">
        <property name="ERROR_NOTICE" value="Message delivery failed after retries, routing to DLQ"/>
    </log>
    <!-- 初始化Kafka连接，发送到死信主题 -->
    <kafkaTransport.init>
        <bootstrapServers>localhost:9092</bootstrapServers>
        <keySerializerClass>org.apache.kafka.common.serialization.StringSerializer</keySerializerClass>
        <valueSerializerClass>org.apache.kafka.common.serialization.StringSerializer</valueSerializerClass>
    </kafkaTransport.init>
    <kafkaTransport.publishMessages>
        <topic>event_dlq_topic</topic>
    </kafkaTransport.publishMessages>
    <!-- 手动提交Offset，避免重复拉取这条失败消息 -->
    <kafkaTransport.commitOffsets/>
</sequence>

4. 修改响应序列，成功后手动提交Offset #

在event_transmit_out_sequence中添加Offset提交逻辑，确保只有外部系统返回成功时才确认消息：

<?xml version="1.0" encoding="UTF-8"?>
<sequence name="event_transmit_out_sequence" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
    <log level="custom">
        <property name="DELIVERY_STATUS" value="Message sent to external system successfully"/>
    </log>
    <!-- 手动提交Kafka Offset -->
    <kafkaTransport.commitOffsets/>
</sequence>

核心逻辑说明 #

• 手动Offset提交：只有消息成功发送并收到响应后才提交Offset，失败消息会被Kafka重新推送。
• 重试机制：针对外部系统临时不可用的场景自动重试，减少进入死信队列的消息数量。
• 死信队列：兜底处理多次重试失败的消息，避免阻塞正常业务，同时保留消息用于后续排查和人工重发。

内容的提问来源于stack exchange，提问作者dragonmaster