spring-cloud / spring-cloud/spring-cloud-stream

MessageRoutingCallback and Routing-expression not worked for event routing

Open
#2,549 4 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

waiting-on-feedback
Dominant language
Java
Stars
1.1k
Forks
646
Avg merge
2d 3h
Merged PRs (30d)
8

Description

Describe the issue
MessageRoutingCallback or Routing-expression not routing the message to appropriate functions.

Routingfunction responsible for routing the events to the desired function and there were two ways that we tried

  1. By implementing MessageRoutingCallback. It's expected to go to anotherConsumeMessage function if headers consist of the key with the desired value. But it didn't route to the expected function rather it's routed to 'consumeMessage'.

PFB the code

 @Bean
  public MessageRoutingCallback customRouter() {
    return new MessageRoutingCallback() {
      @Override
      public FunctionRoutingResult routingResult(Message<?> message) {
        if (!ObjectUtils.isEmpty(message.getHeaders().get("test-key")) &&
            message.getHeaders().get("test-key").toString().equalsIgnoreCase("consume")) {
          return new FunctionRoutingResult("consumeMessage");
        } else if (!ObjectUtils.isEmpty(message.getHeaders().get("test-key")) &&
            message.getHeaders().get("test-key").toString().equalsIgnoreCase("anotherConsume")) {
          return new FunctionRoutingResult("anotherConsumeMessage");
        }
        return MessageRoutingCallback.super.routingResult(message);
      }
    };
  }
  1. We also tried routing expression as can be seen in the below configuration but instead of anotherConsumeMessage, the message routed to ConsumeMessage. PFB the screenshot to check the content of the message and headers picked up by Kafka binder.
spring:
  cloud:
    stream:
      default-binder: kafka
      kafka:
        binder:
          auto-create-topics: false
          brokers:
            - localhost:29092
          header-mapper-bean-name: kafkaheaderMapperTest
          consumer-properties:
            value.deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      bindings:
        consumeMessage-in-0:
          destination: test-kafka-topic
          group: groupIdConfig
        anotherConsumeMessage-in-0:
          destination: test-kafka-topic
          group: groupIdConfig
    function:
      definition: consumeMessage; anotherConsumeMessage
      routing-expression: "headers['contentType'] == 'application/json' ? 'anotherConsumeMessage' : 'consumeMessage'"
      configuration:
  application:
    name: KafkaStreams

Version of the framework

  1. Spring-Cloud-Stream - 3.2.4
  2. spring-cloud-stream-binder-kafka - 3.2.4
    Expected behavior
    Routingfunction should be picked and should decide the routing of the event to the desired function.
    Screenshots
    You can see the below logs to check the activity that happened on the consumer side and the message details received.
image

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start with the MessageRoutingCallback and routing-expression configuration shown in the issue, then inspect how the Kafka binder maps headers and selects functions for the two bindings. Reproduce the case with the provided consumeMessage and anotherConsumeMessage functions; done means each header or expression result routes the event to the expected function.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, kafka, spring
Domain
backend, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.