From b0eb7594e509cad028f36d4091ca9ba95a33662b Mon Sep 17 00:00:00 2001
From: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com>
Date: Tue, 28 Apr 2026 00:13:00 -0700
Subject: [PATCH 1/2] fix python internal marker replay during reconfiguration

---
 .../core/architecture/managers/tuple_processing_manager.py     | 3 ++-
 amber/src/main/python/core/runnables/main_loop.py              | 1 -
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/amber/src/main/python/core/architecture/managers/tuple_processing_manager.py b/amber/src/main/python/core/architecture/managers/tuple_processing_manager.py
index 3d4c9eee26a..a67949e6717 100644
--- a/amber/src/main/python/core/architecture/managers/tuple_processing_manager.py
+++ b/amber/src/main/python/core/architecture/managers/tuple_processing_manager.py
@@ -33,7 +33,8 @@ def __init__(self):
         self.finished_current: Event = Event()
 
     def get_internal_marker(self) -> Optional[InternalMarker]:
-        return self.current_internal_marker
+        ret, self.current_internal_marker = self.current_internal_marker, None
+        return ret
 
     def get_input_tuple(self) -> Optional[Tuple]:
         ret, self.current_input_tuple = self.current_input_tuple, None
diff --git a/amber/src/main/python/core/runnables/main_loop.py b/amber/src/main/python/core/runnables/main_loop.py
index 7b22e086069..9356542a08c 100644
--- a/amber/src/main/python/core/runnables/main_loop.py
+++ b/amber/src/main/python/core/runnables/main_loop.py
@@ -342,7 +342,6 @@ def _process_ecm(self, ecm_element: ECMElement):
                     StartChannel: self._process_start_channel,
                     EndChannel: self._process_end_channel,
                 }[type(self.context.tuple_processing_manager.current_internal_marker)]()
-                self.context.tuple_processing_manager.current_internal_marker = None
 
     def _send_ecm_to_data_channels(
         self, method_name: str, alignment: EmbeddedControlMessageType

From 05ba9f40252e24da19eef23bd9135a784dd6eb5b Mon Sep 17 00:00:00 2001
From: Yicong Huang <17627829+Yicong-Huang@users.noreply.github.com>
Date: Tue, 28 Apr 2026 00:21:18 -0700
Subject: [PATCH 2/2] test(amber): re-enable source propagation reconfiguration
 case

---
 .../texera/amber/engine/e2e/ReconfigurationSpec.scala       | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala
index c13a5c13aa5..92dfba19de1 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ReconfigurationSpec.scala
@@ -256,11 +256,7 @@ class ReconfigurationSpec
     )
   }
 
-  // Disabled: the workflow hangs after the UDF processes the EndChannel ECM in this
-  // multi-worker (Python source -> Python UDF) propagation case, causing a 1-minute
-  // Await timeout. The single-op reconfigure path and the CSV-source variants pass.
-  // Re-enable once the source-propagation completion path is fixed.
-  "Engine" should "propagate reconfiguration through a source operator in workflow" ignore {
+  "Engine" should "propagate reconfiguration through a source operator in workflow" in {
     val sourceOpDesc = TestOperators.pythonSourceOpDesc(10000)
     val udfOpDesc = TestOperators.pythonOpDesc()
     val code = """