-record(watermark, {offset :: integer(), event_time_ms :: integer()}).