Hi, I just want to ask for a general implementation idea if anyone has run into this. I have a finite dataset of user_id, then i have a streaming of data stream of (user_id, event). what my use case is that if one particular user_id has no event coming in for x minutes, the system will alert. This x minutes also includes if there is no data comes in since the system is up. So for example, user 1 has no events comes in for 3 minutes since last event, it will alert, or user 1 has no events at all for 3 minutes since the system is up, it will alert. What i can’t figure out is this: apparently i need to use user_id to keyBy the data stream, and if I boardcast the dataset as boardcast state, then all the parallelisms has the full dataset, then how can i know which user_id showed up in which parallelism so that it doesn’t need to be triggered. In my mind, It feels to me the way to resolve this is to partition dataset to flink operator, so that user_id in the dataset goes to the same stream operator who handles the same user_id of keyBy. But the question is how I know flink operator is in charge of what range of user_id if I use keyBy. Is there a way to match operator to certain kafka partitions if i assume the data is partitioned by user_key in the kafka as input. I hope what I said make sense to you. And I am looking forward to someone giving me some ideas

You can create a composite source with two actual sources: one producing synthetic events for each user_id and than actual source streaming your real events. Then the handler after keyBy will just create a timer for each event (be it synthetic event from system start out real events from Kafka/whatever) and cancel existing timer for the user (if exists). That way you'll have timeout from either start or real event outer user. Or just join two streams of the events (from your event source and from source that enumerated existing users to produce synthetic events) since you'll likely want to turn on the alert timer for me users and permanently turn them off for removed/archived users.

session window and creating timer manually fundamentally are the same, because session window internally still use timer. (i created session function before, so I know, but anyway that’s trivial stuff) So I want to follow up on The synthetic events, as a high level what you suggest is to do a datastream to dataset join?

Maybe I’m missing something but it seems pretty straightforward… 1. Connect the two datastreams, and key by user_id 2. In a custom KeyedCoProcessFunction, when you get the record for the finite (bounded) set of user_ids, start a timer for 3 minutes. 3. In that same function, when you get the (user_id, event) record, reset the timer for another 3 minutes. This assumes that the only user ids you are about are ones from the bounded set. PS - I’d ask the second question (about “how I know flink operator in in charge…“) separately, to keep each thread clean.

@Ken Krugler i think you are right. The main idea is to join a datastream with a dataset, keyed by user_id. correct? I have not done any streaming join with a table (any finite data set) operation, i need to look more into it. in my original thought, i was thinking periodically reads in the full dataset in each flink operator, in such as keyedprocessFunction, then I run into the issue of not knowing which keys are triggered, which are not triggered.

@Rommel any specific reason you are trying to use flink for this use case? Seems it could be solved via some TSDB with a layer of alerting over it.

@Rommel Either side of the data going into your KeyedCoProcessFunction can be changing. You need to decide what the impact a change is to your “finite” stream. E.g. if a user_id can be dropped, then you’ll need to output a record with that information, so the state in the custom function can be updated.

@Ken Krugler thank you for your pointer, i will look into it. @Ananya Goel sorry for my delayed answer. The example i gave is a simplified version of a service i want to create, which is not an alerting system. I just tried to extract the key problem in my thought process and asked for the community to give me some guidance. Sorry for the confusion