From b30ffd1b4fd8bb9455a750d87ba920599e184d91 Mon Sep 17 00:00:00 2001
From: Dream95 <zhou_8621@163.com>
Date: Sun, 26 Jul 2026 14:43:23 +0800
Subject: [PATCH] [FLINK-40237][cep] Clean up expired NFA state in
 processing-time timers

---
 .../apache/flink/cep/operator/CepOperator.java | 11 +++++++++++
 .../flink/cep/operator/CEPOperatorTest.java    | 18 ++++++++++++++++++
 2 files changed, 29 insertions(+)

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
index 11fe970974da8..1e036e4c7b480 100644
--- 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
@@ -369,6 +369,11 @@ public void onProcessingTime(InternalTimer<KEY, VoidNamespace> timer) throws Exc
 
         // STEP 4
         updateNFA(nfa);
+
+        // In order to remove dangling partial matches.
+        if (nfa.getPartialMatches().size() == 1 && nfa.getCompletedMatches().isEmpty()) {
+            computationStates.clear();
+        }
     }
 
     private Stream<IN> sort(Collection<IN> elements) {
@@ -551,6 +556,12 @@ boolean hasNonEmptyPQ(KEY key) throws Exception {
         return !elementQueueState.isEmpty();
     }
 
+    @VisibleForTesting
+    boolean hasNonEmptyNFAState(KEY key) throws Exception {
+        setCurrentKey(key);
+        return computationStates.value() != null;
+    }
+
     @VisibleForTesting
     int getPQSize(KEY key) throws Exception {
         setCurrentKey(key);
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
index df0da998e9bcc..1c51576cfcfe5 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
@@ -928,6 +928,24 @@ public void testCEPOperatorCleanupProcessingTime() throws Exception {
         }
     }
 
+    @Test
+    public void testCEPOperatorClearsExpiredNFAStateInProcessingTime() throws Exception {
+        CepOperator<Event, Integer, Map<String, List<Event>>> operator = getKeyedCepOperator(true);
+
+        try (OneInputStreamOperatorTestHarness<Event, Map<String, List<Event>>> harness =
+                CepOperatorTestUtilities.getCepTestHarness(operator)) {
+            harness.open();
+            harness.setProcessingTime(1L);
+            harness.processElement(new StreamRecord<>(new Event(26, "start", 1.0)));
+
+            assertTrue(operator.hasNonEmptyNFAState(26));
+
+            harness.setProcessingTime(20L);
+
+            assertFalse(operator.hasNonEmptyNFAState(26));
+        }
+    }
+
     @Test
     public void testCEPOperatorSerializationWRocksDB() throws Exception {
         String rocksDbPath = tempFolder.newFolder().getAbsolutePath();