Hi all,
I have multiple consumers that consumes from the same topic based on a particular header event_type. I've tried many configs but I didn't get it to work. I'm trying to not duplicating the binding configs since they share the same one

this is the configuration I'm using :

spring:
  cloud:
    stream:
      function:
        definition: consumerA;consumerB;consumerC
      bindings:
        default:
          destination: same_topic
          group: same_group
          consumer:
            ackEachRecord: true
            autoCommitOffset: true
            autoCommitOnError: false
      kafka:
        binder:
          brokers: localhost
          defaultBrokerPort: 9092
          configuration:
            auto.offset.reset: earliest
          headerMapperBeanName: kafkaHeaderMapper

with custom KAfkaHeaderMap :

@Configuration
public class KafkaHeaderMapperConfig {


    @Bean(name = "kafkaHeaderMapper")
    public KafkaHeaderMapper customHeaderMapper(){
        DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
        Map<String, Boolean> conf = new HashMap<>();
        conf.put("event_type", true);
        mapper.setRawMappedHeaders(conf);
        mapper.setCharset(StandardCharsets.UTF_8);
        return mapper;
    }

}

And this is my custom MessageRoutingCallback (which doesn't trigger once a message is pushed to the topic) :

public class CustomMessageRoutingCallback implements MessageRoutingCallback {

    public static final String EVENT_TYPE = "event_type";

    @Override
    public String functionDefinition(Message<?> message) {
        return Optional.of(message.getHeaders())
            .filter(headers -> headers.containsKey(EVENT_TYPE))
            .map(messageHeaders -> messageHeaders.get(EVENT_TYPE))
            .map(eventType -> EventTypeToBinding.valueOf((String)eventType))
            .map(EventTypeToBinding::getBinding)
            .orElseThrow(() -> new IllegalStateException("event_type was not recognized !! supported values are " + Arrays.toString(EventTypeToBinding.values())));
    }
}

@Getter
@RequiredArgsConstructor
public enum EventTypeToBinding {
    EVENT_A("consumerA-in-0"),
    EVENT_B("consumerB-in-0"),
    EVENT_C("consumerC-in-0");

    private final String binding;
}
`

And finally the declaration of my beans :

/**
 * Configuration class for all bindings used in the application
 */
@Configuration
public class MessageConfiguration {

    /**
     * {@link MessageRoutingCallback} for mapping to the appropriate
     * binding by checking the {@link CustomMessageRoutingCallback#EVENT_TYPE}
     * @return
     */
    @Bean
    public MessageRoutingCallback messageRoutingCallback() {
        return new CustomMessageRoutingCallback();
    }

    @Bean
    public Consumer<A> consumerA() {
        return a -> doSomething();
    }

    @Bean
    public Consumer<B> consumerB() {
        return b -> doSomething();
    }

    @Bean
    public Consumer<C> consumerC() {
        return c -> doSomething();
    }

I got confused by configuration properties, for example should I use spring.cloud.stream.bindings.<channelName>. or spring.cloud.stream.kafka.bindings.<channelName>. !

what am I missing ?