Spring Cloud Stream Kafka 路由表达式

Question

我有一个使用函数式方法的 spring 云流应用程序。该应用程序使用用于将消息过滤到检查负载的消费者函数的路由表达式。

spring:
  cloud:
    function:
      routing-expression: "payload['type'] == 'Event' ? 'handleEvent' : 'commitIgnoredEvent'"
      definition: functionRouter;handleDlqMessage

此配置给出异常：

org.springframework.messaging.MessageHandlingException: error occurred
in message handler [org.springframework.cloud.stream.
function.FunctionConfiguration$FunctionToDestinationBinder@3aff91a]; nested exception is 
org.springframework.expression.spel.SpelEvaluationException:
EL1027E: Indexing into type 'org.apache.avro.generic.GenericData$Record'
is not supported, failedMessage=GenericMessage

但是，这个使用 header 的配置将不会抛出异常：

spring:
  cloud:
    function:
      routing-expression: "header['type'] == 'Event' ? 'handleEvent' : 'commitIgnoredEvent'"
      definition: functionRouter;handleDlqMessage

生产系统不使用 headers 所以我不能使用第二个选项。 routing-expression 仅适用于 header 吗？是否缺少启用负载检查的配置？

Answer 1

您似乎正试图将其作为地图访问。尝试 payload.type（假设有一个 getType() 方法）。

Spring Cloud Stream Kafka 路由表达式

Spring Cloud Stream Kafka routing expression

java

spring

apache-kafka

spring-kafka

spring-cloud-stream