Контроллер покоя для возврата записей в кафку через пружинную кафку

Question

Для моего демонстрационного приложения я должен создать контроллер отдыха, который будет возвращать сообщение в очереди kafka. Я прочитал справочное руководство по spring-kafka, реализовал конфигурацию потребителя и создал bean-компоненты, как показано ниже.

@Configuration
@EnableKafka
public class ConsumerConfiguration {

    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        // list of host:port pairs used for establishing the initial connections to the Kakfa cluster
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        // allows a pool of processes to divide the work of consuming and processing records
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "trx");

        return props;
    }

    @Bean
    public ConsumerFactory<String, Transaction> transactionConsumerFactory() {
        return new DefaultKafkaConsumerFactory<>(
                consumerConfigs(),new StringDeserializer(),new JsonDeserializer<>(Transaction.class)
        );
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Transaction>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Transaction> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(transactionConsumerFactory());

        return factory;
    }

    @Bean
    public Consumer consumer() {
        return new Consumer();
    }

}

и другой потребитель класса, как показано ниже

public class Consumer {

    private CountDownLatch latch = new CountDownLatch(1);

    public CountDownLatch getLatch() {
        return latch;
    }

    @KafkaListener(topics = "${kafka.topic.name}")
    public void receive(Transaction transaction) {
        latch.countDown();
    }
}

Как я могу теперь реализовать логику для получения транзакции от потребителя при каждом попадании на контроллер.

Заранее спасибо.

Answer 1

Ну, то @KafkaListener производит независимый долгоживущий процесс для потоковой передачи записей из Кафки в обратный вызов. Поскольку вы говорите о событии REST GET, у вас нет выбора, если вы не получите KafkaConsumer от ConsumerFactory и назовите его poll() вручную из метода контроллера.