конфиг подключения divolte к kafka

{
  bootstrap.servers = ${?DIVOLTE_KAFKA_BROKER_LIST}
  client.id = divolte.collector
  client.id = ${?DIVOLTE_KAFKA_CLIENT_ID}

  acks = 1
  retries = 0
  compression.type = lz4
  max.in.flight.requests.per.connection = 1
}

при обрыве соединения получаю

WARN  [NetworkClient]: [Producer clientId=divolte.collector] Connection to node -1 could not be established. Broker may not be available.

пробуем менять

  acks = 0
  retries = 5

туже ошибку но с 0

WARN  [NetworkClient]: [Producer clientId=divolte.collector] Connection to node 0 could not be established. Broker may not be available.

https://kafka.apache.org/documentation.html#acks

я где-то близко, но все еще непонятно, какой же - все таки параметр нужно добавить, чтобы происходил реконект

retries и прочее - всё же касается количества попыток отправить сообщение
у вас же происходит следующее: за определённый таймаут не удалось соединиться с брокером. соответственно, необходимо на стороне уже приложения, которое использует kafka, организовать повторный реконнект к брокеру

Да все сходится, по tcpdump видно 5 попыток retry и дальше divolte замирает. И после даже не пытается соединится с kafka

Ставьте тысячу , пусть долбится 😀

проблема оказывается в кубере
это запись в zoo-keeper

{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT","EXTERNAL":"PLAINTEXT"},"endpoints":["PLAINTEXT://10.233.70.243:9092","EXTERNAL://kafka.domain.com:9092"],"jmx_port":5555,"host":"10.233.70.243","timestamp":"1604057730274","port":9092,"version":4}

а это из пода клиента

telnet kafka 9092
Trying 10.233.57.100...
Connected to kafka.analytic-dev.svc.cluster.local.
Escape character is '^]'.

клиент ищет кафку, но не на том IP

ну и в helm сборке есть такое
## Setting "advertised.listeners" here appends to "PLAINTEXT://${POD_IP}:9092,", ensure you update the domain

подскажите, кто-нибудь сталкивался может.
после метода .join идет groupByKey().windowedBy().reduce()

в REDUCE-changelog топик кладутся сообщения у которых ключ совпадает, но WindowId разный и из-за этого далее по стриму метод .suppress(untilWindowCloses()) выдает дубликаты, а нужно одно финальное состояние

Всем привет, граждане, помогите пожалуйста с настройкой SASL (kafka+zookeeper), настроил zookeeper и kafka, службы стартуют, но в логе zookeeper сыпятся ошибки

[2020-10-30 16:45:12,310] WARN Closing client connection due to SASL authentication failure. (org.apache.zookeeper.server.ZooKeeperServer)
[2020-10-30 16:45:12,310] WARN CancelledKeyException causing close of session 0x2016d622e370002 (org.apache.zookeeper.server.NIOServerCnxn)
[2020-10-30 16:45:13,291] INFO Revalidating client: 0x2016d622e370001 (org.apache.zookeeper.server.quorum.Learner)
[2020-10-30 16:45:13,293] WARN Client failed to SASL authenticate: javax.security.sasl.SaslException: DIGEST-MD5: digest response format violation. Mismatched response. (org.apache.zookeeper.server.ZooKeeperServer)
javax.security.sasl.SaslException: DIGEST-MD5: digest response format violation. Mismatched response.

jaas'ы сделал, server.properties, zookeeper.properties настроил, могу скинуть сюда на верификацию

когда стартую только zookeeper ноды, в логах ошибок нет, запускаю ноды кафки тогда ошибки

KafkaServer {
 org.apache.kafka.common.security.plain.PlainLoginModule required
 username="admin"
 password="admin-secret"
 user_admin="admin-secret";
};

Client {
 org.apache.kafka.common.security.plain.PlainLoginModule required
 username="admin"
 password="admin-secret"
 user_admin="admin-secret";
};

скажите - такой kafka_jaas корректный?

🎃 всем привет!
Вот вам задачка на выходные - поковырять два новых туториала (по многочисленным просьбам в интернете)

как избежать дублирования и сохранить порядок про записи https://kafka-tutorials.confluent.io/message-ordering/kafka.htmlhttps://kafka-tutorials.confluent.io/message-ordering/kafka.html
как посчитать количество сообщений в топике https://kafka-tutorials.confluent.io/how-to-count-messages-on-a-kafka-topic/ksql.html

Всем отличных выходных!!!

Очень кстати, спасибо

👍