А если посылать да ещё и читать, то появляются тонкости

Кто-нибудь сталкивался с такой проблемой?
На брокерах несколько сетевых интерфейсов, есть дефолтные листенеры для внутренней коммуникации, а есть advertised листенеры для внешних клиентов. Иногда брокеры после рестарта начинают спонтанно отдавать клиенту при бутстрепе дефолтный адрес вместо того, который указан в advertised.address. Лечится это так же рестартом. Происходит не часто, но когда случается - сильно аффектит клиентов по понятным причинам. Выглядит все так, будто брокер проигнорировал директиву advertised.address, хотя в логах старта видно что этот параметр он вычитал и с ним все ок.

На тесте не могу воспроизвести проблему, поэтому не получается ее локализовать. В джире апача ничего не нашел по этому поводу. Единственное, что помню, что началаось после того как тюнили num net/io threads и quota.window.num.

Подскажите, это нормально собирать стрим в KTable и селектить туда по необходимости или лучше использовать стандартные базы данных? Данные предполагается хранить 2-3 суток.

Дополню, записей ~20 миллионов, если что)

В сутки? Час?

6-8 миллионов в сутки

сообщения большие (сколько точно не помню 😒) Ну там что то типа xml строк ~500-1000 символов

Я делал доклад на московском джуге в прошлом году кажется про interactive queries

Привет, Vik получилось проверить?

а если часть сервисов в DMZ , часть во внутренней сети, кафку где размещать надо

всё зависит от того, кто из них пишет в кафку

сервис в дмз пишет в кафку, типо адаптер к вненшним сервисам

тогда логично во внутренней сети размещать, если прямого доступа извне к кафке не планируется

ага, спасибо

Всем привет. Сталкивался ли кто-нибудь с такой проблемой. Приклад исправно писал в кафку до вчерашнего дня. После перезагрузки падает с "TransactionalId Id authorization failed". Aclи все остались прежними, права все на месте, TransactionalId на прикладе не поменялся. В логах зукипера вижу множественные записи "unknown reqpath:n/a", а также unknown reqpath:/kafka-acl-extended/prefixed/TransactionalId/myTx " при этом znode есть, просматривается через zookeeper shell

📣 Голосуем, каждый голос важен

https://twitter.com/gamussa/status/1303316412006576130?s=21

Дратути, у меня случилось что-то. у некоторых партиций топика __consumer_offsets Isr: 1 вместо 2х, хотя все три брокера живые, как починить?

когда у меня такие залипания были больше суток, помогал только поброкерный рестарт

Всем доброго времени суток.
У меня есть задача обогащать поток данных в Kafka Streams из различных внешних источников.
Одним из вариантов рассматриваю обогащение потока через Processor API а именно через реализацию интерфейса org.apache.kafka.streams.kstream.Transformer.
В методе init планирую инициализировать и устанавливать коннект к внешнему источнику.
В методе transform​ планирую производить непосредственно обогащение.
В методе close планирую закрывать сессию к внешнему источнику.
В связи с этим есть три вопроса:
 1. На сколько такой подход к обогащению потока работоспособен?
 2. Как правильно передавать в Transformer параметры коннекта (Host, Port, Login, Password)
 3. Какие еще способы обогащения из внешних источников посоветуете?
По 2 пункту хотел передавать свои параметры через ProcessorContext, но это пока не поддерживается, обещают в следующих версиях.
Пока сделал передачу параметров через реализацию интерфейса TransformerSupplier.
Наброски кода тут https://pastebin.com/xfQ4kW7j