Hello guys 👋. I’m a software developer and we are developing an alert system using Flink SQL. We want our users to be able to define alerts for specific metrics. In this particular case, we are developing an alert system for the number of users logged in on a particular account. We have two Postgres databases. One defines the alerts (

alert_definitions

) and the other one represents all users’ status (

user_status

). Because these are continuously updating tables, I used

postgres-cdc

connector like so:

CREATE TABLE user_status_binlog (
  id STRING NOT NULL,
  user_id STRING,
  account_id STRING,
  status STRING,
  deleted BOOLEAN,
  inactive BOOLEAN,
  queue_name STRING,
  team_id STRING,
  updated_at TIMESTAMP(3),
  PRIMARY KEY (id) NOT ENFORCED,
  WATERMARK FOR updated_at AS updated_at
) WITH (
 'connector' = 'postgres-cdc',
 'hostname' = 'postgresql',
 'port' = '5432',
 'username' = 'admin',
 'password' = 'admin',
 'schema-name' = 'public',
 'database-name' = 'user-status',
 'table-name' = 'live_user_status',
 'slot.name' = 'flink',
 'debezium.plugin.name' = 'pgoutput',
 'scan.startup.mode' = 'initial',
 'changelog-mode' = 'upsert'
);


CREATE TABLE alert_definitions(
  alert_id STRING NOT NULL,
  metric_id STRING,
  account_id STRING,
  filters STRING,
  min_value NUMERIC,
  max_value NUMERIC,
  filter_status AS JSON_QUERY(filters, '$.status' RETURNING ARRAY<STRING>),
  filter_ring_groups AS JSON_QUERY(filters, '$.ring_groups' RETURNING ARRAY<STRING>),
  filter_team_ids AS JSON_QUERY(filters, '$.team_ids' RETURNING ARRAY<STRING>),
  created_at TIMESTAMP(6),
  PRIMARY KEY (alert_id) NOT ENFORCED
) WITH (
 'connector' = 'postgres-cdc',
 'hostname' = 'postgresql',
 'port' = '5432',
 'username' = 'admin',
 'password' = 'admin',
 'schema-name' = 'public',
 'database-name' = 'live_sentinel_db',
 'table-name' = 'alert_definition',
 'slot.name' = 'alert_definition',
 'debezium.plugin.name' = 'pgoutput',
 'scan.startup.mode' = 'initial',
 'changelog-mode' = 'upsert'
);

Every time a new user changes their status or a new alert is created, I want the query to run and send the results to a kafka topic. Because my query is an updating table, I created a Kafka sink with connector

upsert-kafka

(although the behaviour I’m looking for is more akin to an appending-only table). My query, which counts the number of users logged in, is as follows:

CREATE TEMPORARY VIEW alert_user_counts AS
SELECT
    alert_id,
    account_id,
    active_user_count,
    min_value,
    max_value
FROM 
    (
        SELECT alert_id, account_id, min_value, max_value, filter_ring_groups, filter_status, filter_team_ids 
        FROM alert_definitions 
        WHERE account_id IS NOT NULL
        AND metric_id = 'count-users-logged-in'
    ) alerts,
    LATERAL (
        SELECT
            COUNT(DISTINCT user_id) AS active_user_count
        FROM user_status_binlog AS us
        WHERE
            us.account_id = alerts.account_id
            AND us.status NOT IN ('offline', 'hidden')
            AND us.deleted = FALSE
            AND us.inactive = FALSE
            AND (COALESCE(CARDINALITY(alerts.filter_status), 0) = 0 OR ARRAY_CONTAINS(alerts.filter_status, us.status))
    );

INSERT INTO notification_sink
SELECT 
    counts.account_id,
    counts.alert_id,
    counts.active_user_count as `value`,
    CASE 
        WHEN (counts.min_value IS NULL OR counts.active_user_count >= counts.min_value) 
         AND (counts.max_value IS NULL OR counts.active_user_count <= counts.max_value) 
        THEN 'VIOLATION'
        ELSE 'NOMINAL'
    END AS `type`,
    CURRENT_TIMESTAMP as event_timestamp
FROM alert_user_counts counts

Although the idea behind this job seems pretty simple, I keep finding issues with my implementation. I needed to use a LATERAL JOIN because it was the only way not to get multiple values for the same alert. Currently it’s working fine except when the

alert_definition

table is updated. If a new alert is created, then a lot of messages are sent to kafka. For example: say the account has 3 users logged in. If a new alert is created for that account, we get 3 messages on the kafka topic, like so:

{"account_id":"account-id","alert_id":"10","value":1,"type":"VIOLATION","event_timestamp":"2025-12-18 18:08:30.991"}
{"account_id":"account-id","alert_id":"10","value":2,"type":"VIOLATION","event_timestamp":"2025-12-18 18:08:30.992"}
{"account_id":"account-id","alert_id":"10","value":3,"type":"VIOLATION","event_timestamp":"2025-12-18 18:08:30.993"}

I don’t really understand this ‘counting’ behaviour. It’s as if the user_status table was an appending only table and this new alert is joining with multiple past versions. Can someone please explain to me why this behaviour is happening? P.S. You might realize that this implementation doesn’t really work when no users are logged in (In which case it should return a 0 but Flink SQL doesn’t return any lines). I’m aware of that, but right now it’s not where I’m focusing on.