Flink GlobalWindow Trigger обрабатывает только событие триггера ⇐ JAVA

[Anonymous]

У меня есть ключ потока данных по свойству события, которое затем передается в глобальное окно и активируется при возникновении определенного события. Проблема в том, что когда окно запускается для обработки событий, оно обрабатывает только событие триггера. Вот код:

Код: Выделить всё

public class MyEvent {
public String key;
public String value;

public MyEvent() {
// Default constructor
}

public MyEvent(String key, String value) {
this.key = key;
this.value = value;
}

public String getKey() {
return key;
}

public void setKey(String key) {
this.key = key;
}

public String getValue() {
return value;
}

public void setValue(String value) {
this.value = value;
}

@Override
public String toString() {
return "Key: " + key + ", Value: " + value;
}
}

public class KeyedGlobalWindowTriggerExample {

public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream input = env.socketTextStream("localhost", 9091)
.map(new MapFunction() {
@Override
public MyEvent map(String value) {
// Assuming the input stream is in the format "key,value"
String[] parts = value.split(",");
return new MyEvent(parts[0], parts[1]);
}
});

// Key By Event Property
KeyedStream keyedStream = input
.keyBy(event -> event.key);

//Create a Custom Trigger
keyedStream.window(GlobalWindows.create())
.trigger(new Trigger() {
@Override
public TriggerResult onElement(MyEvent event, long timestamp, GlobalWindow window, TriggerContext ctx) {
if ("eod".equals(event.getKey())) {
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
}

@Override
public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) {
return TriggerResult.CONTINUE;
}

@Override
public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) {
return TriggerResult.CONTINUE;
}

@Override
public void clear(GlobalWindow window, TriggerContext ctx) {
// Handle clearing of window state if necessary
}
})
.process(new MyProcessWindowFunction())
.print();

env.execute("Keyed Global Window Trigger Example");
}
}

public class MyProcessWindowFunction extends ProcessWindowFunction {
@Override
public void process(String key, Context context, Iterable elements, Collector out) {
for (MyEvent element : elements) {
out.collect(element.toString());
}
}
}

Как убедиться, что он обрабатывает все события в глобальном окне с помощью триггерного события?
Если я отправлю следующие события:
test_one, one
event_two, two
event_three, Three
event_four, four
eod, eod
Вывод:
/>Ключ: eod, События: [Ключ: eod, Значение: eod, ]
Я ожидал увидеть все события, отправленные в это окно.

Подробнее здесь: https://stackoverflow.com/questions/787 ... gger-event