Hello, What could be the reasons of watermark not be propagating downstream ?

can you explain more? it appears you have watermarks in your temporal join

Hi Jeremy, thanks for looking into this I am using the Ververica

postgres-cdc

connector on a Temporal Join, but I noted the Watermark is not propagating downstream. Please see on the image how the Watermark on the first operator is greater than the next operator and it's not being updated.

CREATE TABLE presence_feature (
    serial STRING NOT NULL,
    remoteMac STRING NOT NULL,
    version STRING NOT NULL,
    tsMs TIMESTAMP(3) NOT NULL,
    features ARRAY<DOUBLE NOT NULL> NOT NULL,
    WATERMARK FOR tsMs AS tsMs - INTERVAL '8' SECOND
) WITH (
    'connector' = 'kafka',
    ...
)

CREATE TABLE group_thing_association (
    group_id VARCHAR NOT NULL,
    thing_serial VARCHAR NOT NULL,
    operation_ts TIMESTAMP(3) METADATA FROM 'op_ts' VIRTUAL,
    WATERMARK FOR operation_ts AS operation_ts,
    PRIMARY KEY(thing_serial) NOT ENFORCED
) WITH (
    'connector' = 'postgres-cdc',
    ...
)

SELECT
    '0.1.0' AS version,
    group_id AS resourceId,
    'group' AS resourceType,
    MAX(tsMs) AS tsMs,
    MAX(status) AS status
FROM (
    SELECT
        serial,
        remoteMac,
        HOP_ROWTIME(tsMs, interval '5' MINUTE, interval '15' MINUTE) AS tsMs,
        presence_predict(features, tsMs, serial, remoteMac) AS status
    FROM presence_feature
    GROUP BY HOP(tsMs, interval '5' MINUTE, interval '15' MINUTE), serial, remoteMac
)
JOIN group_thing_association FOR SYSTEM_TIME AS OF tsMs
    ON serial = thing_serial
GROUP BY TUMBLE(tsMs, INTERVAL '5' MINUTE), group_id

what version of flink are you running this on?

1.16

when running this you have new data coming in both sources or just the one?

the first source ( above ) send data every 5 mins but the second source ( bellow ) send data only when a row is modified on the Postgres table I read that an operator with 2 sources, it get the minimum watermark of its 2 sources, which most of the time is the second source Is there any way to set the operator to get the maximum watermark of the 2 sources ? For now, I did a workaround by modifying a row in the Postgres table every few seconds. By doing that, the second source watermark get incremented. thanks Jeremy