Создать нового производителя от потребителя Кафки?

Question

Как создать новый Kafka Producer из существующего Consumer с Java?

Answer 1

Я рекомендую использовать библиотеку Kafka Streams.Он читает данные из тем Кафки и выполняет некоторую обработку и записывает их в другие темы.

Это может быть проще для вас.https://kafka.apache.org/documentation/streams/

Текущее ограничение: кластер источника и назначения должен совпадать с потоками Kafka.В противном случае вам нужно использовать Processor API для определения другого целевого кластера.

Другой подход - просто определить производителя в программе-потребителе.Везде, где ваше правило соответствует (основано на смещении или любых условиях), вызывайте producer.send() метод

Answer 2

Вы не можете создать KafkaProducer из экземпляра KafkaConsumer.

Вы должны явно создать KafkaProducer, используя те же параметры подключения, что и у вашего производителя.

Учитывая упомянутый вами вариант использования (копирование данных из одной темы в другую), я бы рекомендовал использовать Kafka Streams. На самом деле в Кафке есть пример, который делает именно это: https://github.com/apache/kafka/blob/trunk/streams/examples/src/main/java/org/apache/kafka/streams/examples/pipe/PipeDemo.java