Confluent Platform не может правильно отправлять / потреблять сообщения в теме Kafka - PullRequest
Новая
       5

Confluent Platform не может правильно отправлять / потреблять сообщения в теме Kafka

0 голосов
/ 25 августа 2018

В настоящее время я использую Maven с io.fabric8 docker-maven-plugin, чтобы автоматически раскручивать Kafka и ZooKeeper. Это моя текущая конфигурация, которая отлично работает: 

<properties>
    <zookeeper.port>2181</zookeeper.port>

    <kafka.host>127.0.0.1</kafka.host>
    <kafka.port>9092</kafka.port>
</properties>

<plugin>
    <groupId>io.fabric8</groupId>
    <artifactId>docker-maven-plugin</artifactId>
    <version>${docker-maven-plugin.version}</version>

    <configuration>
        <showLogs>true</showLogs>
        <images>

            <image>
                <name>wurstmeister/zookeeper:${zookeeper.version}</name>
                <alias>zookeeper</alias>
                <run>
                    <ports>
                        <port>${zookeeper.port}:2181</port>
                    </ports>
                </run>
            </image>

            <image>
                <name>wurstmeister/kafka:${kafka.version}</name>
                <alias>kafka</alias>
                <run>
                    <ports>
                        <port>${kafka.port}:9092</port>
                    </ports>
                    <links>
                        <link>zookeeper:zookeeper</link>
                    </links>
                    <env>
                        <KAFKA_ADVERTISED_HOST_NAME>${local.ip}
                        </KAFKA_ADVERTISED_HOST_NAME>
                        <KAFKA_ADVERTISED_PORT>${kafka.port}</KAFKA_ADVERTISED_PORT>
                        <KAFKA_ZOOKEEPER_CONNECT>zookeeper:2181</KAFKA_ZOOKEEPER_CONNECT>
                        <KAFKA_MESSAGE_MAX_BYTES>15000000</KAFKA_MESSAGE_MAX_BYTES>
                    </env>
                </run>
            </image>

            <image>
                <name>confluentinc/cp-ksql-server:5.0.0</name>
                <alias>cp-ksql-server</alias>
                <run>
                    <ports>
                        <port>8088:8088</port>
                    </ports>
                    <links>
                        <link>kafka:kafka</link>
                    </links>
                    <env>
                        <KSQL_BOOTSTRAP_SERVERS>${local.ip}:9092</KSQL_BOOTSTRAP_SERVERS>
                        <KSQL_LISTENERS>http://0.0.0.0:8088/</KSQL_LISTENERS>
                        <KSQL_KSQL_SERVICE_ID>confluent_test_2</KSQL_KSQL_SERVICE_ID>
                    </env>
                </run>
            </image>

        </images>

    </configuration>
</plugin>

Сейчас мне нужно больше гибкости и инструментов для мониторинга, чтобы можно было правильно проанализировать мой локальный экземпляр Kafka, и именно поэтому я решил начать использовать All-In-One Confluent Platform https://docs.confluent.io/current/quickstart/ce-docker-quickstart.html

Я выполнил шаги, описанные в документации выше, и смог запустить Confluent Platform с помощью команды docker-compose up -d на следующей docker-compose.yml https://github.com/confluentinc/cp-docker-images/tree/master/examples/cp-all-in-one. После этого я смог получить доступ к Центру управления Confluent на http://localhost:9021

Мое приложение также запускается, но сообщения не отправляются или не используются должным образом. Это журнал: 

2018-08-25 20:54:33.786  INFO 11304 --- [nio-8080-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka version : 1.1.0
2018-08-25 20:54:33.786  INFO 11304 --- [nio-8080-exec-1] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : fdcf75ea326b8e07
2018-08-25 20:54:33.794  INFO 11304 --- [ad | producer-1] org.apache.kafka.clients.Metadata        : Cluster ID: NCbZ7abTTdu6SQHhthMbIw
2018-08-25 20:54:33.827  INFO 11304 --- [nio-8080-exec-1] c.p.domain.service.post.PostServiceImpl  : Message sent to the post creation queue: Post [id=5b8197d9e67db62c28ebe8d5, status=PENDING, externalPostId=2bdbd597-cc30-4478-b946-e873c8d53eec, chatName= ....]

2018-08-25 20:55:03.875 ERROR 11304 --- [ad | producer-1] o.s.k.support.LoggingProducerListener    : Exception thrown when sending a message with key='null' and payload='Post [id=5b8197d9e67db62c28ebe8d5, status=PENDING, externalPostId=2bdbd597-cc30-4478-b946-e873c8d53e...' to topic post.send:

org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for post.send-0: 30047 ms has passed since batch creation plus linger time

Я вижу свои темы там:

enter image description here

Что там может быть не так и как это исправить?

ОБНОВЛЕНО

Я включил журналы отладки на Kafka в своем приложении и сейчас вижу следующую ошибку: 

java.io.IOException: Can't resolve address: broker:9092
    at org.apache.kafka.common.network.Selector.doConnect(Selector.java:235) ~[kafka-clients-1.1.0.jar:na]
    at org.apache.kafka.common.network.Selector.connect(Selector.java:214) ~[kafka-clients-1.1.0.jar:na]
    at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:793) [kafka-clients-1.1.0.jar:na]
    at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:230) [kafka-clients-1.1.0.jar:na]
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.sendEligibleCalls(KafkaAdminClient.java:792) [kafka-clients-1.1.0.jar:na]
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1002) [kafka-clients-1.1.0.jar:na]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_171]
Caused by: java.nio.channels.UnresolvedAddressException: null
    at sun.nio.ch.Net.checkAddress(Net.java:101) ~[na:1.8.0_171]
    at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:622) ~[na:1.8.0_171]
    at org.apache.kafka.common.network.Selector.doConnect(Selector.java:233) ~[kafka-clients-1.1.0.jar:na]
    ... 6 common frames omitted

1 Ответ

0 голосов
/ 26 августа 2018

Не удается разрешить адрес: брокер: 9092

Кафка возвращает клиентам тех слушателей, которых знает о 

KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:29092

Как правило, bootstrap.servers нужен только один адрес, но остальные адреса кластера Kafka будут возвращены клиенту для установления соединений с отдельными брокерами.

Когда вы подключаетесь за пределами сети Docker к localhost:9092, тогда переадресация порта Docker будет работать, но затем Kafka возвращается к вашему клиенту broker:9092, и клиент отключается, потому что вы не настроили DNS между вашим хостом и сеть Docker (и вам это не нужно, если вы не используете Docker Swarm).

Ваше решение будет два варианта

  1. Используйте localhost:29092 в своем коде, и все может начать работать лучше
  2. Запустите ваш Java-код в контейнере Docker в той же сетевой конфигурации с помощью файла Compose и оставьте ссылку kafka:9092 (это похоже на работу ваших контейнеров KSQL и Control Center)
...