Flink pipeline with firing results on event

I have stream of objects with address and list of organizations:

@Data
class TaggedObject {
    String address;
    List<String> organizations;
}

Is there a way to do the following using apache flink:

1. Merge organization lists for objects with same address
2. Send all results to Sink when some event occurs. E.g. when user sends control message to a kafka topic or another DataSource
3. Keep all objects for future accumulations

I tried using global window and custom trigger:

public class MyTrigger extends Trigger<TaggedObject, GlobalWindow> {
    @Override
    public TriggerResult onElement(TaggedObject element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {
        if (element instanceof Control) return TriggerResult.FIRE;
        else return TriggerResult.CONTINUE;
    }

But it seems to give only Control element as a result. Other elements were ignored.

If you want a generic control signal that triggers output for ALL addresses, then you'll need to use a broadcast stream. You combine your stream of addresses with your control stream and then perform the appropriate logic (merging organizations for an address, or triggering output) inside of your custom implementation of a KeyedBroadcastProcessFunction.

It seems like you should just key the stream by address and then use a KeyedProcessFunction (with a List- or MapState) to store the different organizations. Then as soon as an event comes in, you can just output the entries of the State.

Kind Regards
Dominik