GlobalWindow with custom tigger doesn't work correctly

[Daniel Krenn]

Hello people!

I have a DataStream, which has events with with a continuing number 
which signifies their belonging to a production cycle. In essence, this 
is what the data looks like:

value, production cycle
12.0, 2000
12.3, 2000 one production cylce
12.2, 2000

0.0, 2001
0.4, 2002 another production cycle
1.1, 2002

55.0, 2003
60.0, 2003 another production cycle
70.0, 2003

I have to do some calculations over the events of each production cycle. 
I want to use Flink's window API for that. This is how I'm doing it 
right now:

DataStream<String> test = 
streamExecEnv.readTextFile("C:/Projects/Python/testdata.txt")
    .map(newImaginePaperDataConverterTest()) // convert data to POJO
    .assignTimestampsAndWatermarks(newImaginePaperAssigner()) // Assign 
timestamps for event time
    .keyBy((ImaginePaperData event) ->event.lunum) //<- the production 
cycle number
    .window(GlobalWindows.create()) // create global window
    .trigger(newLunumTrigger()) // "split" the window with a custom 
trigger
    .process(newImaginePaperWindowReportFunction()); // apply a function 
over the aggregated events

I'm getting a "DataStream" out of a text file, just for testing 
purposes. The problem is that what I'm doing only aggregates one single 
event for a production cycle. Why is that? I thought keying the stream 
by the production cycle number already partitions the stream anyways. 
The trigger says when the production cycle number is changed, a new 
global window is started and the events of the current window are 
aggregated. What am I missing here?
Just to be safe, here is my implementation of the custom trigger:

publicclassLunumTriggerextendsTrigger<ImaginePaperData, GlobalWindow> {

privatestaticfinallongserialVersionUID = 1L;

publicLunumTrigger() {}

privatefinalValueStateDescriptor<Integer> prevLunum = 
newValueStateDescriptor<>("lunum", Integer.class);

@Override
publicTriggerResultonElement(ImaginePaperDataelement, longtimestamp, 
GlobalWindowwindow, TriggerContextctx) throwsException {

ValueState<Integer> lunumState = ctx.getPartitionedState(prevLunum);

if (lunumState.value() == null || 
!(element.lunum.equals(lunumState.value()))) {
System.out.println("LUNUM BEFORE: " + lunumState.value() + " NEW LUNUM: 
" + element.lunum + " ==> FIRE!");
lunumState.update(element.lunum);
returnTriggerResult.FIRE_AND_PURGE;
}

System.out.println("LUNUM BEFORE: " + lunumState.value() + " NEW LUNUM: 
" + element.lunum + " ==> HOLD!");
lunumState.update(element.lunum);
returnTriggerResult.CONTINUE;
}

@Override
publicTriggerResultonProcessingTime(longtime, GlobalWindowwindow, 
TriggerContextctx) throwsException {
returnTriggerResult.CONTINUE;
}

@Override
publicTriggerResultonEventTime(longtime, GlobalWindowwindow, 
TriggerContextctx) throwsException {
returnTriggerResult.CONTINUE;
}

@Override
publicvoidclear(GlobalWindowwindow, TriggerContextctx) throwsException {
ctx.getPartitionedState(prevLunum).clear();
}
}

I'm very grateful for your help.

Regards,

Daniel