Интеграция Kafka с RabbitMQ

Question

Я хочу объединить Kafka и RabbitMQ,

Я могу отправлять сообщения в RabbitMQ, но сообщение не используется.Статус канала показывает статус - 0 в консоли RabbitMQ.

Изменение пользователя как администратора, но все та же проблема.

Развертывание Kafka через Confluent (а не индивидуальное развертывание Zookeeper Kafka и т. Д.) Итакже установлен разъем Rabbitmq через Confluent-hub.

Пожалуйста, помогите.

[2019-03-26 06:39:19,151] ERROR Consumer io.confluent.connect.rabbitmq.ConnectConsumer@454774b5 (amq.ctag-Unaj3jmbQQctolAwNzU2SQ) method handleDelivery for channel AMQChannel(amqp://guest@0:0:0:0:0:0:0:1:5672/,1) threw an exception for channel AMQChannel(amqp://guest@0:0:0:0:0:0:0:1:5672/,1) (com.rabbitmq.client.impl.ForgivingExceptionHandler:124)
java.lang.NullPointerException
at io.confluent.connect.rabbitmq.MessageConverter.basicProperties(MessageConverter.java:127)
at io.confluent.connect.rabbitmq.SourceRecordBuilder.sourceRecord(SourceRecordBuilder.java:40)
at io.confluent.connect.rabbitmq.ConnectConsumer.handleDelivery(ConnectConsumer.java:69)
at com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:149)
at com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:100)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

Answer 1

Я проверял RabbitMQ Kafka Connector и обнаружил точно такую ​​же ошибку.

Сообщение AMQP состоит из двух частей.Заголовок и содержание сообщения.Я только предоставлял содержимое сообщения, а не заголовок.

Вы должны создать объект BasicProperties с метаданными.Документы Java можно просмотреть здесь .

Затем при публикации в RabbitMQ с использованием метода basicPublish используйте объект BasicProperties, который вы создали ранее, в качестве одного из аргументов.

Документы Java для basicPublish можно найти здесь .

Надеюсь, это поможет