Spring-Integration - kafka Outbound-Adapter, der keine Themenwerte als Spring-Bean freigibt

Question

Ich habe erfolgreich einen Kafka-Outbound-Channel-Adapter mit festem Topic-Namen integriert. Nun möchte ich den Themennamen konfigurierbar machen und ihn daher über Anwendungseigenschaften verfügbar machen.

application.properties enthalten eine der folgenden Eintrag:

kafkaTopic:testNewTopic 

sieht Meine Konfigurationsklasse wie unten:

@Configuration 
@Component 
public class KafkaConfig { 

    @Value("${kafkaTopic}") 
    private String kafkaTopicName; 

    @Bean 
    public String getTopic(){ 
    return kafkaTopicName; 
    } 

    @Bean 
    public KafkaTemplate<String, String> kafkaTemplate() { 
     return new KafkaTemplate<>(producerFactory()); 
    } 

    @Bean 
    public ProducerFactory<String, String> producerFactory() { 
     Map<String, Object> props = new HashMap<>(); 
     props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");//this.brokerAddress); 
     props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); 
     props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); 

     // set more properties 
     return new DefaultKafkaProducerFactory<>(props); 
    } 
} 

und in meinem si-config.xml, habe ich folgendes verwendet (ex: topic = "getTopic"):

<int-kafka:outbound-channel-adapter 
     id="kafkaOutboundChannelAdapter" kafka-template="kafkaTemplate" 
     auto-startup="true" sync="true" channel="inputToKafka" topic="getTopic"> 
    </int-kafka:outbound-channel-adapter> 

Die Konfiguration ist jedoch nicht in der Lage, das Thema Nam abzuholen e bei Belichtung über Bean. Aber es funktioniert gut, wenn ich den Wert des Themennamens hart kodiere.

Kann mir bitte jemand vorschlagen, was ich hier falsch mache?

Funktioniert topic innerhalb Kafka Outbound-Kanal den Wert als Bean bezeichnet?

Wie ich es als jede Anwendung externalisieren kann mein Dienstprogramm unterschiedlichen kafka Themennamen

Answer 1

Das topic Attribut ist für String-Wert liefern.

jedoch unterstützt Eigenschaft Platzhalter Auflösung:

topic="${kafkaTopic}" 

und auch Spel Bewertung für oben genannte Bohne:

topic="#{getTopic}" 

Nur weil diese durch die XML-Parser-Konfiguration erlaubt ist.

Sie können jedoch darauf achten, dass KafkaTemplate, die Sie in die <int-kafka:outbound-channel-adapter> haben defaultTopic Eigenschaft. Daher müssen Sie sich nicht um dieses XML kümmern.

Und eine weitere Option für Sie ist die Spring Integration Annotations Konfiguration. Wo Sie eine @ServiceActivator für die KafkaProducerMessageHandler@Bean definieren können:

@ServiceActivator(inputChannel = "inputToKafka") 
@Bean 
KafkaProducerMessageHandler kafkaOutboundChannelAdapter() { 
    kafkaOutboundChannelAdapter adapter = new kafkaOutboundChannelAdapter(kafkaTemplate()); 
    adapter.setSync(true); 
    adapter.setTopicExpression(new LiteralExpression(this.kafkaTopicName)); 
    return adapter; 
}