How to set a seperate bootstrap server to a DLT of a binding

user_1758194 0
Viewed 278

I need to send consumed invalid messages into a DLT but in a separate bootstrap server. I currently have this config:

spring.cloud.stream.binders.some-kafka-binder.type=kafka
spring.cloud.stream.binders.some-kafka-binder.environment.spring.cloud.stream.kafka.binder.brokers=localhost:29092

spring.cloud.stream.bindings.processor-in-0.binder=some-kafka-binder
spring.cloud.stream.bindings.processor-in-0.group=${spring.application.name}
spring.cloud.stream.bindings.processor-in-0.destination=outbound-topic
spring.cloud.stream.kafka.bindings.processor-in-0.consumer.configuration.value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
spring.cloud.stream.kafka.bindings.processor-in-0.consumer.configuration.schema.registry.url=${schema.registry.url}
spring.cloud.stream.kafka.bindings.processor-in-0.consumer.configuration.specific.avro.reader=true
spring.cloud.stream.kafka.bindings.processor-in-0.consumer.enable-dlq=true
spring.cloud.stream.kafka.bindings.processor-in-0.consumer.dlq-name=outbound-topic.DLT
spring.cloud.stream.kafka.bindings.processor-in-0.consumer.dlq-producer-properties.configuration.value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
spring.cloud.stream.kafka.bindings.processor-in-0.consumer.dlq-producer-properties.configuration.schema.registry.url=${schema.registry.url}
spring.cloud.stream.kafka.bindings.processor-in-0.consumer.dlq-producer-properties.configuration.bootstrap.servers=localhost:9092

...but I'm getting this error:

Caused by: java.lang.IllegalStateException: bootstrap.servers cannot be overridden at the binding level; use multiple binders instead
        at org.springframework.util.Assert.state(Assert.java:76)
        at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.getProducerFactory(KafkaMessageChannelBinder.java:560)
        at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.getErrorMessageHandler(KafkaMessageChannelBinder.java:1148)
        at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.getErrorMessageHandler(KafkaMessageChannelBinder.java:158)
        at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.registerErrorInfrastructure(AbstractMessageChannelBinder.java:695)
        at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.registerErrorInfrastructure(AbstractMessageChannelBinder.java:639)
        at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:734)
        at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder.createConsumerEndpoint(KafkaMessageChannelBinder.java:158)
        at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:408)
        ... 27 common frames omitted

Is there a way in spring cloud I can set this up? I really wish that there's no need for me to do a custom DLT implementation.

If ever you're asking why there's a need for another bootstrap setup for the DLT: there are 2 separate AWS KMS accounts involve.

spring-kafka spring-cloud-stream spring-cloud apache-kafka spring-boot
1 Answers

When you enable DLT, the destination needs to be on the same Kafka cluster as the input topic. At the moment, the binder does not provide a way to use a different cluster separately for DLT. I think you can do a workaround though if you are willing to create an extra topic. Let the DLT be on the first cluster so that when an error occurs it goes to the DLT on the same cluster from where the data came from. Then you can define another passthrough function for which the DLT is the input topic and the output topic is configured with the binder for the second cluster.

@Bean
public Consumer<IN> consumer() {

// records in error go to a DLT.
}

// input is the DLT topic from first cluster
// output is a topic on the second cluster.
@Bean
public Function<IN, OUT> function() {

  return in -> out;

}

spring.cloud.stream.binders.some-kafka-binder-1.type=kafka
spring.cloud.stream.binders.some-kafka-binder-1.environment.spring.cloud.stream.kafka.binder.brokers=localhost:29092
spring.cloud.stream.binders.some-kafka-binder-2.type=kafka
spring.cloud.stream.binders.some-kafka-binder-2.environment.spring.cloud.stream.kafka.binder.brokers=localhost:9092

Also, remove the bootstrap config from the DLT configuration.

Feel free to add a new feature request at the GitHub repository for Kafka binder. We can consider using maybe the StreamBridge API to accommodate this use case.

user_20708610
Related