Spring Cloud Stream - One To Many Transformation

user_12449845 0
Viewed 321

I have a simple Spring Cloud Stream application where the stream accepts a raw JSON string and produces a list of Objects.

Version used :

<spring.cloud.version>Hoxton.SR2</spring.cloud.version>

Signature of stream used:

@Bean
public Function<String, List<EmailEvent>> normalizeStream() {
  return value -> {
    return getNormalizedEvents(value);
  };
}

Expectation is that although the output signature of the stream is a List, it should be serialized as individual items in the queue. This change is perfectly working in this version of Cloud stream.

But when I upgrade to Hoxton.SR4 or above, this expectation is breaking and the producer is producing as a list only instead of individual items.

Any help on what I might be missing would be greatly appreciated.

spring-kafka spring-cloud-stream list
1 Answers

I mean that the thing is in the method signature that you're using.

With this version the results are as expected. Check examples below.

<parent>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
    <version>3.1.0-RC1</version>
</parent>

1st CASE. One to Many (wrapped in a list)

In this first example, each String message produces a Message with a List<String> Object Payload that is sent to kafka.

@Bean
    public Function<String, List<String>> oneToMany() {
        return str -> {
            return List.of(str, "custom");
        };
    }

Reading output topic with kafkacat:

 % Reached end of topic one-to-many [0] at offset 180
["a","custom"]

2ns CASE. One to Many (one by one)

In this other example (I think that is what you're looking for) each string is converted to a List and then each item in that List<Message<<String>> is sent to kafka by the binding bridge.

@Bean
    public Function<String, List<Message<String>>> oneToMany() {
        return str -> {
            return List.of(MessageBuilder.withPayload(str).build(), MessageBuilder.withPayload("custom").build());
        };
    }

Reading output topic with kafkacat:

 % Reached end of topic one-to-many [0] at offset 185
a
custom
user_146315970
Related