Передача динамического значения Кафке потребителей темы во время выполнения для сообщений приводится адаптер канала

Question

Требование заключается в установлении тем атрибут ниже во время выполнения без перезапуска сервера .Как мы можем достичь здесь. В настоящее время мы читаем значение из файла свойств, но здесь требуется перезагрузка сервера, чтобы отразить сделанные изменения.

пример: sample.properties (внутри каталога развертывания)

topic.list = topic1, topic2

и хотели бы потреблять от topic3 ​​в будущем без перезагрузки сервера.

ПРИМЕЧАНИЕ: найти что темы является окончательным переменной.

попытался прочитать ключ (topic.list) из пути к файловой системе (вне каталога развертывания), но не повезло.

Любое предложение.

<int-kafka:message-driven-channel-adapter 

       id="inAdapter" 
       channel="fromKafka" 
       connection-factory="connectionFactory" 
       key-decoder="kafkaKeyDecoder" 
       payload-decoder="kafkaDecoder"        
       topics="${topic.list}" 
       offset-manager="offsetManager"/> 

Answer 1

Вы можете использовать DSL Java динамически добавлять адаптеры для дополнительных тем по запросу ...

@Autowired 
private IntegrationFlowContext flowContext; 

public void addAnotherListenerForTopics(String... topics) { 
    IntegrationFlow flow = 
     IntegrationFlows.from(Kafka.messageDrivenChannelAdapter(consumerFactory(), topics)) 
      .channel("fromKafka") 
      .get(); 
    this.flowContext.registration(flow).register(); 
} 

и

bean.addAnotherListenerForTopics("added.new"); 

ПОМ:

<dependency> 
    <groupId>org.springframework.integration</groupId> 
    <artifactId>spring-integration-java-dsl</artifactId> 
    <version>1.2.1.RELEASE</version> 
</dependency> 

Примечание что если вы используете брокерский раздел assi gnment, новый контейнер нуждается в другом идентификаторе группы, чтобы избежать отмены существующих назначений.