Это их фреймворк делает routing. Kafka Streams так не может из коробки

А по фреймворку может кто подсказать? У меня в конфиге все то же самое, за исключением, что используется только Function. На выход берет адреса брокеров из kafka1 биндера, а должен из kafka2.

rdkafka не умеет коннектор, соответственно и гошка тоже

Мне кажется, что Spring Cloud Stream Kafka Binder не умеет из коробки обрабатывать такие кейсы.

Вот цитата из документации
«It continues to remain hard to robust error handling using the high-level DSL; Kafka Streams doesn’t natively support error handling yet.»

https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/2.2.1.RELEASE/spring-cloud-stream-binder-kafka.html#_error_handling

Но я вот не пойму даже как это можно сделать только с помощью кафка стримс. Можно задать три обработчика ошибок. DeserializationExceptionHandler, ProductionExceptionHandler и UncaightExceptionHandler. В первые два я не попадаю, а третий когда ловит, то в нем есть только возможность сделать stream.stop(); System.exit() или stream.start(), но после перезапуска обработка начинается с того же оффсета.

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

кто-то юзал  MM2 на предмет миграции Kafka Streams приложений?
например, такой use case:
1) имеем X KS приложений, которые надо поочередно мигрировать из кластера A в кластер В
2) для каждого приложения:
2.1) зераклим входные и внутрение топики из А в В, где они получают новые имена (Tn => B.Tn).
полагаемся, что офсеты консюмеров зеркалятся автоматически.
2.2) в В создаем выходные топики B.TO и зеркалим их обратно в A (B.TOn => TOn). Здесь возникает вопрос что префикс A на обратном зеркалировании надо убирать, вроде можно через конфиги ?
2.3) останавливаем наш апп в А и рестартуем в В. тут другой вопрос - как менять конфигурацию для В что бы мы подхватили новые имена топиков (с префиксами) и реплицировнные офсеты консюмеров - и то и другое зависит от application.id  
3) в принципе, для 2.1 можно реплицировать без префикса (как для 2.2). так, наверное, будет даже проще.  это уже не будет active-active, но в моем случае это не так важно. к сожалению, не смог найти какой-либо конткретики для мигации KS приложений, в основном все про low level. непонятно почему нигде не обсуждается специфика KS, тк ньюансы и вопросы (например, то что KS приложение это Producer и Consumer в одном флаконе, и разделить их никак)  очевидны ...

Вот в том то и проблема, что ошибка при сериализации во внутренний стейтстор при выполнении join.

а конкретнее, почему возникает ошибка, разбирались ? возможная причина - несовместимое изменение формата результата, если так то это надо решать репроцессингом

да, причина ошибки ясна. конкретно это попытка сеарелизовать BigDecimal co scale=0 в авро тип, который требует scale=2.

Понятно, что это нужно чинить и ловить на тестах.

Но  вопрос в том, как защитить себя от такой ошибки на проде, при которой из-за одного битого сообщения возможна ситуация, когда приложение полностью блокируется и нет возмжности  продолжить его работу из-за того что оно не может пропустить такое сообщение.

Тесты серде в авро и обратно, перед деплоем?

Не отловил все комбинации данных и из за одного сообщеня откат на проме?)

Критикуешь-предлагай))

Как вариант обязательные ревью схемы данных

Кто-то же решил scale=0 поставить

Это все костыли и косвенные вещи, проблему они не решают, а откладывают

Да верно. но ведь тесты не 100 %   гарантия  что не будет ошибки.


Например, если мы пишем  рест сервис и прилетел запрос с некорректным форматом. Ок - отбили его и работаем дальше. А как быть со стримами

Я бы предложил, но мы с Лехой в одной команде 😁😁

Нужно пропускать такие сообщения? Или как?

Хотя бы