Apache Camel处理错误时未删除SQS消息的问题

你的问题核心是：虽然错误处理逻辑触发了日志输出，但SQS消息仍被反复投递，没有按预期删除。这是因为SQS组件的deleteAfterRead=true仅在Exchange完全成功（无残留异常标记）时才会删除消息，而当前的错误处理虽然用.handled(true)标记了异常已处理，但可能存在Exchange异常状态未被彻底清除的情况，导致SQS判定处理失败。

以下是具体解决步骤：

1. 确认异常被正确捕获 #

首先检查JSON缺失必填字段时抛出的具体异常类型：

• 若使用Kotlin+Jackson，缺失非空字段时会抛出MissingKotlinParameterException，它是JsonMappingException的子类，默认会被你的onException列表捕获，但如果未生效，可显式添加该异常类：

  .onException(
      InvalidPayloadException::class.java,
      JsonParseException::class.java,
      JsonMappingException::class.java,
      MissingKotlinParameterException::class.java, // 新增Kotlin特有的缺失参数异常
      ExpressionEvaluationException::class.java,
      IllegalStateException::class.java
  )
  

2. 彻底清除Exchange异常标记 #

在错误处理链中添加.clearException()，确保Exchange的异常状态被完全清除，让SQS识别处理成功：

class ParseErrorHandlingRouteConfiguration : RouteConfigurationBuilder() {
    override fun configuration() {
        routeConfiguration("parseErrorHandling")
            .onException(
                // 异常列表保持不变或补充MissingKotlinParameterException
                InvalidPayloadException::class.java,
                JsonParseException::class.java,
                JsonMappingException::class.java,
                MissingKotlinParameterException::class.java,
                ExpressionEvaluationException::class.java,
                IllegalStateException::class.java
            )
                .useOriginalMessage()
                .handled(true)
                .clearException() // 新增：彻底清除异常标记
                .maximumRedeliveries(0)
                .log(LoggingLevel.ERROR, "Parse/mapping error [route=\${routeId}, exchange=\${exchangeId}]: \${exception.message}")
                .to("seda:dlq.parse")
            .end()
    }
}

3. 验证useOriginalMessage()的影响 #

如果上述调整后仍无效，可暂时移除.useOriginalMessage()测试。该方法会将当前消息替换为原始消息，可能干扰Exchange的状态判定。若移除后消息正常删除，可调整错误处理逻辑（比如无需保留原始消息，或通过其他方式缓存原始消息）。

4. 入口路由添加兜底错误处理 #

若前三种方式都不生效，可在SQS入口路由中添加全局错误处理，确保所有异常都被标记为已处理：

class ContentChangedIngestRoute : RouteBuilder() {
    override fun configure() {
        from("aws2-sqs://myQueue?deleteAfterRead=true")
            .routeId("content-changed-ingest")
            .onException(Exception::class.java)
                .handled(true)
                .log(LoggingLevel.ERROR, "Ingest route error: \${exception.message}")
            .end()
            .to("direct:content-changed")
    }
}

注意：此方式会捕获所有异常，需确保与原有错误处理逻辑无冲突。

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