spring-cloud / spring-cloud/spring-cloud-stream
MessageRoutingCallback and Routing-expression not worked for event routing
Nobody has claimed this yet.
- 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
- 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);
}
};
}
- 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
- Spring-Cloud-Stream - 3.2.4
- 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.
Contributor guide
No contributing guide indexed for this repository
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- 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