+ немного допилить strimzi для этого варианта потребуется…

Возьмите обычный ACL из коробки, настройте использование секьюрного транспорта и аутентификацию в ZK. Должно вам быть достаточным.

Большое спасибо!

Товарищи! Всем привет! А есть такое понятие у кафки как "оьратная связь" по пакету?, что то типа ACK3. Т.е. продюсер должен знать что сделал консюмер с пакетом (успешно прочитал или не прочитал)?

Нет такого

Тут же pub/sub

Можно быть и несколько потребителей

Понял теперь спасибо!

А зачем продюсеру знать что сделал консьюмер?

Запилили обмен по кафке на 1С. Возникла ситуация, когда консьюмер принял пакет с ошибкой. Продюсер эту ошибку должен хотя бы узнать

Просто 0 или 1, примитив

Spring-kafka умеет обрабатывать все ошибки. Может там поискать

Спасибо!! Буду искать

Кто-нибудь может объяснить разницу между:

MyObject myObject = getMyObject();
Message message = MessageBuilder
     .withPayload(myObject)
     .setHeader(KafkaHeaders.TOPIC, replyTopic)
     .setHeader(KafkaHeaders.CORRELATION_ID, correlationId)
     .build();
kafkaTemplate.send(message).addCallback(...)

и

MyObject myObject = getMyObject();
kafkaTemplate.send(String replyTopic, myObject);

Проблема в следующем: при получении сообщения в первом случае в кафка листенере объект автоматически десериализуется @Payload MyObject myObject, а во втором случае - ошибка, что невозможно десериализовать и в стектрейсе массив байт.

Так это нужно язык указывать наверное

java

А что внутри getMyObject? И, объявлен ли класс MyObjectкак Serializable?

getMyObject() - условный метод, который возвращает объект с данными (не писать же всю бизнес-логику)).
да, объект MyObject implements Serializable
айдишник внутри присутствует, с этим все нормально.
разница только в методах отправки сообщения через кафку

во втором случае вот такая ошибка (основная часть)

37:o.s.kafka.listener.LoggingErrorHandler : Error while processing: ConsumerRecord(topic = my-topic-name, partition = 0, offset = 154, CreateTime = 1624860278866, serialized key size = 0, serialized value size = 97, headers = RecordHeaders(headers = [RecordHeader(key = springDeserializerExceptionValue, value = [-84, -19, 0, 5, 115, 114, 0, 69, 111, 114, 103, 46, 115, 112, 114, 105, 110, 103, 102, 114, 97, 109, 101, 119, 111, 114, 107, 46, 107, 97, 102, 107, 97, 46, 115, 117, 112, 112, 111, 114, 116, 46, 115, 101, 114, 105, 97, 108, 105, 122, 101, 114, 46, 68, 101, 115, 101