[FLINK-37168][python] Clean up TimerRegistrationAction in unregisteredTimers list after timers are registered (#26011)

Author: suez1224

flink-python/src/main/java/org/apache/flink/streaming/api/operators/python/process/timer/TimerRegistrationAction.java

@@ -18,6 +18,9 @@
 
 package org.apache.flink.streaming.api.operators.python.process.timer;
 
+import java.util.List;
+
+/** {@link TimerRegistrationAction} used to register Timer. */
 public class TimerRegistrationAction {
 
     private final TimerRegistration timerRegistration;
@@ -26,17 +29,33 @@ public class TimerRegistrationAction {
 
     private boolean isRegistered;
 
+    private final List<TimerRegistrationAction> containingList;
+
     public TimerRegistrationAction(
-            TimerRegistration timerRegistration, byte[] serializedTimerData) {
+            TimerRegistration timerRegistration,
+            byte[] serializedTimerData,
+            List<TimerRegistrationAction> containingList) {
         this.timerRegistration = timerRegistration;
         this.serializedTimerData = serializedTimerData;
+        this.containingList = containingList;
         this.isRegistered = false;
     }
 
     public void run() {
+        registerTimer();
+        cleanup();
+    }
+
+    public void registerTimer() {
         if (!isRegistered) {
             timerRegistration.setTimer(serializedTimerData);
             isRegistered = true;
         }
     }
+
+    private void cleanup() {
+        if (isRegistered) {
+            containingList.remove(this);
+        }
+    }
 }

flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonFunctionRunner.java

@@ -195,7 +195,7 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
 
     private transient Environment environment;
 
-    private transient List<TimerRegistrationAction> unregisteredTimers;
+    private transient volatile List<TimerRegistrationAction> unregisteredTimers;
 
     public BeamPythonFunctionRunner(
             Environment environment,
@@ -311,7 +311,7 @@ public void open(ReadableConfig config) throws Exception {
                 ShutdownHookUtil.addShutdownHook(
                         this, BeamPythonFunctionRunner.class.getSimpleName(), LOG);
 
-        unregisteredTimers = new LinkedList<>();
+        unregisteredTimers = Collections.synchronizedList(new LinkedList<>());
     }
 
     @Override
@@ -352,10 +352,12 @@ public void process(byte[] data) throws Exception {
 
     @Override
     public void drainUnregisteredTimers() {
-        for (TimerRegistrationAction timerRegistrationAction : unregisteredTimers) {
-            timerRegistrationAction.run();
+        synchronized (unregisteredTimers) {
+            for (TimerRegistrationAction timerRegistrationAction : unregisteredTimers) {
+                timerRegistrationAction.registerTimer();
+            }
+            unregisteredTimers.clear();
         }
-        unregisteredTimers.clear();
     }
 
     @Override
@@ -703,7 +705,9 @@ private TimerReceiverFactory createTimerReceiverFactory() {
                 (timer, timerData) -> {
                     TimerRegistrationAction timerRegistrationAction =
                             new TimerRegistrationAction(
-                                    timerRegistration, (byte[]) timer.getUserKey());
+                                    timerRegistration,
+                                    (byte[]) timer.getUserKey(),
+                                    unregisteredTimers);
                     unregisteredTimers.add(timerRegistrationAction);
                     environment
                             .getMainMailboxExecutor()