Всем привет! Поделитесь опытом, пожалуйста. У нас на проекте используется Кафка, клиент на Java.
Есть задача отправлять эвенты клиентам по Server Sent Event. После подключения клиент должен получить все эвенты пока он был оффлайн, а затем продолжать получать сообщения, созданные после подключения.
Хочу реализовать следующим образом. Продюсеры, которые хотят уведомить клиента, отправляют сообщения в специальный топик. Когда клиент подключается, создаем нового Кафка консьюмера на основе id пользователя(каждый пользователь это отдельная консьюмер-группа), подписываемся на все партишены в топике. Затем запрашиваем батч сообщений из Кафки, отфильтровываем сообщения для клиента, отправляем их по SSE, затем фиксируем оффсет в Кафка и так до бесконечнеости. Если клиент отключается, то просто отписываемся от всех партишенов. Простой тест показал, что при 2 партишенах прочитать 1500 сообщений пользователя из 3 млн сообщений в топике одним консьюмером занимает примерно 1,5 - 2 мин, что не проходит по требованиям. Поэтому хочу увеличить количество портишенов(например, до 1024) У нас продюсеры  роутят сообщения по ключу. Консьюмер будет делать предположение в каком партишене искать сообщения на основе ключа и подписываться только на этот партишен.  

Есть такие вопросы:
1. Насколько плохо иметь большое количество консьюмеров с точки зрения Кафки. Допустим, их будет параллельно более 1000 на этот топик?
2. Вывезет ли Кафка такое количество партишенов и консьюмеров?
3. Есть в описанной схеме что-то что я не учел или можно сделать лучше?
4. В схеме с фиксированным количеством партишенов есть проблема, что нельзя будет их увеличить потом так, чтобы не потерять сообщения пользователя. Как бы это можно было бы решить?

я бы рассматривал кафку как промежуточный буфер и потоково сортировал нотификации, которые бы раскладывал в редис по ключам (user:natofication.type)

всем привет, хотел узнать, есть ли какая-то возможность обработать ошибку о том, что не удалось подключиться к кафке, когда она пишет Connection to node could not be established. Broker may not be available.? Судя по сорцам, там нет никаого эксепшна и никак это обработать нельзя, в гугле тоже ничего не нашел, кроме нескольких сообщений о том, что так и задумано

sasl plain

запускаю такой вот KafkaContainer (от testcontainers)

kafkaContainer.withEnv("KAFKA_LISTENERS", "SASL_PLAINTEXT://:9092");
kafkaContainer.withEnv("KAFKA_ADVERTISED_LISTENERS", "SASL_PLAINTEXT://localhost:9092");
kafkaContainer.withEnv("KAFKA_ZOOKEEPER_CONNECT", "zookeeper:2181");
kafkaContainer.withEnv("KAFKA_OPTS", "-Djava.security.auth.login.config=./kafka_server_jaas.conf");
kafkaContainer.withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "SASL_PLAINTEXT");
kafkaContainer.withEnv("KAFKA_SECURITY_PROTOCOL", "SASL_PLAINTEXT")
kafkaContainer.withEnv("KAFKA_SASL_MECHANISMS", "PLAIN")
kafkaContainer.withEnv("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "PLAIN")
kafkaContainer.withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "PLAINTEXT") // !!!!!

получаю вот такое:
org.apache.kafka.common.errors.IllegalSaslStateException: Unexpected handshake request with client mechanism PLAINTEXT, enabled mechanisms are []

оно ведь включено...

забавно то, что конфиг 1в1 с docker-compose, в котором это все ходит

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

забавно то, что конфиг 1в1 с docker-compose, в котором это все ходит

что за композ?

тык https://gist.github.com/cristaloleg/2c1e035d8884c4aacd1f7f62bf9fa4b0

TC Не использует wurstmeister
TC используюет правильный образ от Confluent

хмм

Такое да