[Flink 32701] [cep] Fix CEP Operator Memory Leak Issue (apache#24084)

viktorsomogyi · Mar 21, 2024 · a9cde49 · a9cde49
1 parent 3f4a809
commit a9cde49
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 4 deletions.
diff --git a/...libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBuffer.java b/...libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBuffer.java
@@ -272,6 +272,11 @@ void advanceTime(long timestamp) throws Exception {
                 iterator.remove();
             }
         }
+
+        // memory leak resolution
+        if (eventsCount.isEmpty()) {
+            eventsCount.clear();
+        }
     }
 
     EventId registerEvent(V value, long timestamp) throws Exception {

diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java
@@ -326,6 +326,11 @@ public void onEventTime(InternalTimer<KEY, VoidNamespace> timer) throws Exceptio
 
         // STEP 4
         updateNFA(nfaState);
+
+        // In order to remove dangling partial matches.
+        if (nfaState.getPartialMatches().size() == 1 && nfaState.getCompletedMatches().isEmpty()) {
+            computationStates.clear();
+        }
     }
 
     @Override

diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStateAccessTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/nfa/NFAStateAccessTest.java
@@ -81,8 +81,8 @@ public void testComplexBranchingAfterZeroOrMore() throws Exception {
         nfaTestHarness.consumeRecords(inputEvents);
 
         assertEquals(58, sharedBuffer.getStateReads());
-        assertEquals(33, sharedBuffer.getStateWrites());
-        assertEquals(91, sharedBuffer.getStateAccesses());
+        assertEquals(41, sharedBuffer.getStateWrites());
+        assertEquals(99, sharedBuffer.getStateAccesses());
     }
 
     @Test
@@ -149,7 +149,7 @@ public boolean filter(Event value, Context<Event> ctx)
         nfaTestHarness.consumeRecords(inputEvents);
 
         assertEquals(90, sharedBuffer.getStateReads());
-        assertEquals(31, sharedBuffer.getStateWrites());
-        assertEquals(121, sharedBuffer.getStateAccesses());
+        assertEquals(34, sharedBuffer.getStateWrites());
+        assertEquals(124, sharedBuffer.getStateAccesses());
     }
 }