У меня есть конвейер потока данных Goolge, написанный с использованием Apache beam [java], который потребляет данные из облачной подписки Google. Входные данные отображаются в окне сеанса:
inputMessages.apply("WindowTransform", Window.into(Sessions.withGapDuration(Duration.standardMinutes(5)))
.withAllowedLateness(Duration.standardDays(7))
.triggering(
AfterPane.elementCountAtLeast(1)
)
.discardingFiredPanes());
Проблема в том, что все работает нормально для начального слива, но как только все сообщения из подписки потребляются, конвейер не будет потреблять никаких новых сообщений. Также нет сообщений об ошибках. Мне пришлось вручную повторно отправить задание для обработки сообщений снова, но оно снова перестает обрабатывать новые сообщения, которые не были частью первоначальной утечки.
Что могло быть причиной такого поведения?