Ну не всегда прям flatten. В тестах flatten, в проде можно как-то иначе ждать второго F. Главное - дожидаться обоих.

В проде я на стриме делаю через parEvalMap

Т.е. дожидаешься каждый батч конкурентно?

Я боюсь, что если я начну дожидаться у меня перф деграднется.

Я в кафку посылаю то, что ждать не обязательно

Конкретно, я делаю вот так:

???.evalMap(msg.publish).parEvalMap(maxPublishBufferSize)(identity)

msg.publish на самом деле ничего не паблишит. Он только добавляет сообщение в буфер на паблишинг. Это внешний F. Сделующим шагом я дожидаюсь внутреннего F — он отрабатывает, когда данные из буфера успешно отправлены в кафку. Чтобы не ждать каждого сообщения (то есть не делать буфер размером 1) я делаю parEvalMap(maxPublishBufferSize)(identity), где maxPublishBufferSize — достаточно большое число. В кафка клиенте уже и так есть куча настроек этого буффера, поэтому мне нет смысла (обычно) ещё и на приложении делать какое-то ограничение буфера. Поэтому maxPublishBufferSize я выбираю заведомо больше, чем может влезть в буфер. И получается, что когда буффер отправляется в кафку, сразу пачка этих внутренних F внутри parEvalMap резолвится и идёт дальше по стриму.
В моих приложениях обычно у меня на входе сообщения из кафки. И я не могу не ждать успешного паблишинга. Потому что мне после паблишинга нужно делать коммит.

По идее, на продакшене и старая и новая версии должны работать идентично. Если тебе не нужно ждать, когда реально залилось в кафку (ты готов терять сообщения при отправке в кафку), то fire and forget семантика должна сохраниться.

Интересно еще то, что flatten как будто форсирует исполнение ио в файбере. С флэттеном посылка тестового батча занимает 15-20 сек, без флэттена вообще ничего не происходит с очень долгим таймаутом.

На прошлой имплементации через прямой сенд с деферредом все за долю секунды делалось.

Используется c.е.IO с глобал ЕС.

Семантика вроде не поменялась, а вот работа ранлупа сильно аффектит всё.

Попробуй явно заранить внутренний F и явно задискардить его

Типа map(_.void). void

А что значит явно заранить? Он и так является .join'ом от какого-то файбера

Надо еще поковыряться с c.e.IO, он только в тестах. Похоже у него фейрнес совсем на дне, м.б. надо пошифтить вручнуб

Ну вообще да, может и в этом дело

Какие гарантии есть в ce2 после вызова unsafeRunSync относительно исполнения файберов, если мейн ИО завершилось?

Если их не джоинить

Кажется в IOPlatform где-то шатдаунились скедулеры

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

Насколько я знаю, нормально будет только в ce3