Query and Possible Bug Report for Flink CEP's Management of NFAState

[Puneet Duggal <pu...@gmail.com>]

Hi,

I am writing this email to bring attention to a potential issue I have
encountered while working with the Complex Event Processing (CEP) library
in Flink v1.17.


The CEP Operator maintains a keyed state called NFAState, which holds two
queues: one for partial matches and one for completed matches. When a key
is first encountered, the CEP creates a starting computation state and
stores it in the partial matches queue. As more events occur that match the
defined conditions (e.g., a TAKE condition), additional computation states
get added to the queue, with their specific type (normal, pending, end)
depending on the pattern sequence.

However, I have noticed that the starting computation state remains in the
partial matches queue even after the pattern sequence has been completely
matched. This is also the case for keys that have already timed out. As a
result, the state gets stored for all keys that the CEP ever encounters,
leading to a continual increase in the checkpoint size.


*This is how one can reproduce this.*

Pattern Sequence - A not_followed_by B within 5 mins


On my local machine, I started this pipeline and started sending events at
the rate of 10 events per second (only A) and as expected after 5 mins, CEP
started sending pattern matched output with the same rate. But the issue
was that after every 2 mins (checkpoint interval), checkpoint size kept on
increasing. Expectation was that after 5 mins (2-3 checkpoints), checkpoint
size will remain constant since any window of 5 mins will consist of the
same number of unique keys (older ones will get matched or timed out hence
removed from state). But as you can see below attached images, checkpoint
size kept on increasing till 40 checkpoints (around 1.5hrs).


P.S. - After 3 checkpoints (6 mins), the checkpoint size was around 1.78MB.
Hence assumption is that ideal checkpoint size for a 5 min window should be
less than 1.78MB.


[image: Screenshot 2023-07-26 at 11.45.06 AM.png]


[image: Screenshot 2023-07-26 at 11.50.28 AM.png]


As you can see after 39 checkpoints, I triggered a savepoint for this
pipeline. After that I used a savepoint reader to investigate what all is
getting stored in CEP states. Below code investigates *NFAState *of
CEPOperator for potential memory leak.

public void testNfaStateReader() throws Exception {
> PatternRule patternRule = TestUtils.getResource("AnfbB/Pattern.json", new
> TypeReference<PatternRule>(){});
> StreamExecutionEnvironment env = EnvironmentUtility.
> configureStreamEnvironment(patternRule, 1, TimeCharacteristic.EventTime);
> System.out.println(((FsStateBackend)env
> .getStateBackend()).getCheckpointPath());
> SavepointReader savepointReader =
> SavepointReader.read(env,
> "file:///tmp/flink/savepoints/savepoint-765510-37636a00002f", new
> FsStateBackend("file:///tmp/flink/checkpoints"));
> DataStream<NFAStateOutput> stream = savepointReader.readKeyedState(
> OperatorIdentifier.forUid(SELECT_PATTERN_EVENTS), new NFAStateReaderTest
> .NFAStateReaderFunction());
> stream.print();
> env.execute();
> }
>

>
static class NFAStateReaderFunction extends KeyedStateReaderFunction<
> DynamicTuple, NFAStateOutput> {
>
> private ValueState<NFAState> computationStates;
> private static Long danglingNfaCount = 0L;
> private static Long newNfaCount = 0L;
> private static Long minTimestamp = Long.MAX_VALUE;
> private static Long minKeyForCurrentNfa = Long.MAX_VALUE;
> private static Long minKeyForDanglingNfa = Long.MAX_VALUE;
> private static Long maxKeyForDanglingNfa = Long.MIN_VALUE;
> private static Long maxKeyForCurrentNfa = Long.MIN_VALUE;
>
> @Override
> public void open(Configuration parameters) {
> computationStates = getRuntimeContext().getState(new
> ValueStateDescriptor<>(NFA_STATE_NAME, new NFAStateSerializer()));
> }
>
> @Override
> public void readKey(DynamicTuple key, Context ctx, Collector<
> NFAStateOutput> out) throws Exception {
> NFAState nfaState = computationStates.value();
> System.out.println("**************************************************");
>
> if (Objects.requireNonNull(nfaState.getPartialMatches().peek()).getStartTimestamp()
> != -1) {
> minTimestamp = Math.min(minTimestamp, nfaState
> .getPartialMatches().peek().getStartTimestamp());
> minKeyForCurrentNfa = Math.min(minKeyForCurrentNfa, Long.parseLong
> (key.getTuple().getField(0)));
> maxKeyForCurrentNfa = Math.max(maxKeyForCurrentNfa, Long.parseLong
> (key.getTuple().getField(0)));
> newNfaCount++;
> } else {
> danglingNfaCount++;
> minKeyForDanglingNfa = Math.min(minKeyForDanglingNfa, Long.parseLong
> (key.getTuple().getField(0)));
> maxKeyForDanglingNfa = Math.max(maxKeyForDanglingNfa, Long.parseLong
> (key.getTuple().getField(0)));
> }
> NFAStateOutput nfaStateOutput =
> new NFAStateOutput(
> danglingNfaCount,
> minTimestamp,
> newNfaCount,
> minKeyForCurrentNfa,
> maxKeyForCurrentNfa,
> minKeyForDanglingNfa,
> maxKeyForDanglingNfa);
> out.collect(nfaStateOutput);
> }
> }
>

>
 @Data
> @NoArgsConstructor
> @AllArgsConstructor
> static class NFAStateOutput implements Serializable {
> private Long danglingNfaCount;
> private Long minTimestamp;
> private Long newNfaCount;
> private Long minKeyForCurrentNfa;
> private Long maxKeyForCurrentNfa;
> private Long minKeyForDanglingNfa;
> private Long maxKeyForDanglingNfa;
> }
>

As an output it printed nfaStateOutput for each key but since all the
attributes in nfaStateOutput are aggregates, hence finalOutput printed was

NFAStateReaderTest.NFAStateOutput(danglingNfaCount=34391,
> minTimestamp=1690359951958, newNfaCount=3000, minKeyForCurrentNfa=6244230,
> maxKeyForCurrentNfa=6247229, minKeyForDanglingNfa=629818,
> maxKeyForDanglingNfa=6244229)
>

As we can see, checkpoint is storing approximately 34391 dangling states
(for keys which have expired (matched or timed out) ) whereas there are
only 3000 active keys (for which there are partial matches which are
eligible for further pattern sequence matching).

Hence, I am curious about the reasoning behind this design choice,
specifically why the starting state remains in the partial matches queue
for all keys, even those that have either timed out or completed their
matches. Additionally, I am wondering what the implications would be if we
were to delete this starting state assuming that

   1. it is the only state left in the partial match queue.
   2. completed match queue in nfa state is empty.