是否可以使用 Apache Flink 处理 RabbitMQ 中的中毒消息？答案

【问题标题】：Is it possible to handle poisoned messages in RabbitMQ with Apache Flink?是否可以使用 Apache Flink 处理 RabbitMQ 中的中毒消息？
【发布时间】：2021-11-16 12:26:18
【问题描述】：

我正在创建一个简单的 Apache Flink 作业（使用 Scala），它只尝试打印表示 RabbitMQ 队列接收到的事件的案例类 (RMQSource)

我创建了自己的反序列化模式（使用 Jackson），当使用的消息实际上是表示案例类的 JSON 时，它可以正常工作。但是，如果队列接收到格式错误的事件（我猜我们可以称其为“中毒消息”），作业将失败并继续重新启动。我必须清除队列，然后作业状态变为“正在运行”。

问题：

当收到有毒消息时，如何防止作业失败？我可以在使用之前验证消息吗？如果我可以在 Rabbit 中设置死信交换，我应该在哪里（如果可能）代表 Apache Flink 作为消费者进行否定确认？有更好的方法来处理这个问题并使用下一个格式正确的消息保持作业运行？

我的自定义 DeserializationSchema 提供给 RMQSource[Test]：

class eventSerializationSchema extends  DeserializationSchema[Test] {

  @throws(classOf[IOException])
  def deserialize(message: Array[Byte]): Test =  objectMapper.readValue(message, classOf[Test])

  def isEndOfStream(nextElement: Test): Boolean = false

  def getProducedType: TypeInformation[Test] = createTypeInformation[Test]
}

object eventSerializationSchema{

  val objectMapper: ObjectMapper = new ObjectMapper().registerModule(DefaultScalaModule)

}

中毒消息到达消费队列时出现错误：

com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'a': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
 at [Source: (byte[])"a"; line: 1, column: 2]
    at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:2337)
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:720)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3593)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._handleUnexpectedValue(UTF8StreamJsonParser.java:2688)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._nextTokenNotInObject(UTF8StreamJsonParser.java:870)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:762)
    at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4684)
    at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4586)
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3609)
    at org.angoglez.deserializers.eventSerializationSchema.deserialize(eventSerializationSchema.scala:17)
    at org.angoglez.deserializers.eventSerializationSchema.deserialize(eventSerializationSchema.scala:14)
    at org.apache.flink.streaming.connectors.rabbitmq.RMQDeserializationSchemaWrapper.deserialize(RMQDeserializationSchemaWrapper.java:47)
    at org.apache.flink.streaming.connectors.rabbitmq.RMQSource.processMessage(RMQSource.java:319)
    at org.apache.flink.streaming.connectors.rabbitmq.RMQSource.run(RMQSource.java:331)
    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:116)
    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:73)
    at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:323)

【问题讨论】：

标签： scala rabbitmq apache-flink

【解决方案1】：

在我看来，您有几个选择：

捕获反序列化方法中抛出的异常，并删除有害记录。
捕获反序列化方法中抛出的异常，并以某种方式将您想了解的有关这些有害记录的信息编码到正在生成的对象中。然后在下游处理函数中，过滤掉这些毒记录，并将它们发送到侧输出。
不要在反序列化器中应用 ObjectMapper，而是在可以将毒记录直接发送到侧输出的链式处理函数中进行真正的反序列化。

【讨论】：