В настоящее время я читаю pub / sub и указываю, какой атрибут использовать в качестве временной метки
PubsubIO.readMessagesWithAttributes().withTimestampAttribute("timestamp")
Однако, если я передаю сообщение без этого атрибута временной метки, мой процесс неограниченной потоковой передачисбои.Это проблема, потому что кто-то может случайно опубликовать сообщение для проверки и сбоя всего потока данных.Я мог бы фильтровать по этому атрибуту, но тогда я не знаю, какую функцию использовать, чтобы установить «отметку времени» в качестве основного атрибута отметки времени.
Кто-нибудь знает, как установить атрибут отметки времени после фильтрации?