A workflow with two python udf operators doesn't complete the execution.

[kunwp1]

What happened?

A workflow has 1-out python udf and python udf operator. 1-out python udf operator produces 0 tuples. In this case, the downstream python udf operator doesn't complete the execution.

How to reproduce?

Run this workflow.

Untitled workflow (13).json

Branch

main

Commit Hash (Optional)

No response

What browsers are you seeing the problem on?

No response

Relevant log output

2026-05-24 16:40:36.928 | ERROR    | threading:run:1012 - An error has been caught in function 'run', process 'MainProcess' (38104), thread 'main_loop_thread' (6156333056):
Traceback (most recent call last):

  File "/Users/kunwoopark/.pyenv/versions/3.12.9/lib/python3.12/threading.py", line 1032, in _bootstrap
    self._bootstrap_inner()
    │    └ <function Thread._bootstrap_inner at 0x109d77060>
    └ <Thread(main_loop_thread, started 6156333056)>
  File "/Users/kunwoopark/.pyenv/versions/3.12.9/lib/python3.12/threading.py", line 1075, in _bootstrap_inner
    self.run()
    │    └ <function Thread.run at 0x109d76d40>
    └ <Thread(main_loop_thread, started 6156333056)>
> File "/Users/kunwoopark/.pyenv/versions/3.12.9/lib/python3.12/threading.py", line 1012, in run
    self._target(*self._args, **self._kwargs)
    │    │        │    │        │    └ {}
    │    │        │    │        └ <Thread(main_loop_thread, started 6156333056)>
    │    │        │    └ ()
    │    │        └ <Thread(main_loop_thread, started 6156333056)>
    │    └ <bound method StoppableQueueBlockingRunnable.run of <core.runnables.main_loop.MainLoop object at 0x115e50320>>
    └ <Thread(main_loop_thread, started 6156333056)>

  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/util/stoppable/stoppable_queue_blocking_thread.py", line 65, in run
    self.receive(self.interruptible_get())
    │    │       │    └ <function StoppableQueueBlockingRunnable.interruptible_get at 0x10fe774c0>
    │    │       └ <core.runnables.main_loop.MainLoop object at 0x115e50320>
    │    └ <function MainLoop.receive at 0x115dcb9c0>
    └ <core.runnables.main_loop.MainLoop object at 0x115e50320>

  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/runnables/main_loop.py", line 148, in receive
    match(
    └ <function match at 0x1100884a0>

  File "/Users/kunwoopark/.pyenv/versions/3.12.9/lib/python3.12/site-packages/pampy/pampy.py", line 299, in match
    return run(action, lambda_args)
           │   │       └ [ECMElement(tag=ChannelIdentity(from_worker_id=ActorVirtualIdentity(name='Worker:WF321-PythonUDFSourceV2-operator-bf7723af-dc...
           │   └ <bound method MainLoop._process_ecm of <core.runnables.main_loop.MainLoop object at 0x115e50320>>
           └ <function run at 0x1100734c0>
  File "/Users/kunwoopark/.pyenv/versions/3.12.9/lib/python3.12/site-packages/pampy/pampy.py", line 48, in run
    return action(*var)
           │       └ [ECMElement(tag=ChannelIdentity(from_worker_id=ActorVirtualIdentity(name='Worker:WF321-PythonUDFSourceV2-operator-bf7723af-dc...
           └ <bound method MainLoop._process_ecm of <core.runnables.main_loop.MainLoop object at 0x115e50320>>

  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/runnables/main_loop.py", line 340, in _process_ecm
    {

  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/runnables/main_loop.py", line 291, in _process_end_channel
    self.complete()
    │    └ <function MainLoop.complete at 0x115dca480>
    └ <core.runnables.main_loop.MainLoop object at 0x115e50320>

  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/runnables/main_loop.py", line 100, in complete
    self.context.state_manager.transit_to(WorkerState.COMPLETED)
    │    │       │             │          │           └ WorkerState.COMPLETED
    │    │       │             │          └ <enum 'WorkerState'>
    │    │       │             └ <function StateManager.transit_to at 0x115da9260>
    │    │       └ <core.architecture.managers.state_manager.StateManager object at 0x115d37b90>
    │    └ <core.architecture.managers.context.Context object at 0x115e50380>
    └ <core.runnables.main_loop.MainLoop object at 0x115e50320>

  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/architecture/managers/state_manager.py", line 73, in transit_to
    raise InvalidTransitionException(
          └ <class 'core.architecture.managers.state_manager.InvalidTransitionException'>

core.architecture.managers.state_manager.InvalidTransitionException: Cannot transit from READY to COMPLETED
Exception in thread main_loop_thread:
Traceback (most recent call last):
  File "/Users/kunwoopark/.pyenv/versions/3.12.9/lib/python3.12/threading.py", line 1075, in _bootstrap_inner
    self.run()
  File "/Users/kunwoopark/.pyenv/versions/3.12.9/lib/python3.12/threading.py", line 1012, in run
    self._target(*self._args, **self._kwargs)
  File "/Users/kunwoopark/.pyenv/versions/3.12.9/lib/python3.12/site-packages/loguru/_logger.py", line 1251, in catch_wrapper
    return function(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/util/stoppable/stoppable_queue_blocking_thread.py", line 65, in run
    self.receive(self.interruptible_get())
  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/runnables/main_loop.py", line 148, in receive
    match(
  File "/Users/kunwoopark/.pyenv/versions/3.12.9/lib/python3.12/site-packages/pampy/pampy.py", line 299, in match
    return run(action, lambda_args)
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/kunwoopark/.pyenv/versions/3.12.9/lib/python3.12/site-packages/pampy/pampy.py", line 48, in run
    return action(*var)
           ^^^^^^^^^^^^
  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/runnables/main_loop.py", line 340, in _process_ecm
    {
  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/runnables/main_loop.py", line 291, in _process_end_channel
    self.complete()
  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/runnables/main_loop.py", line 100, in complete
    self.context.state_manager.transit_to(WorkerState.COMPLETED)
  File "/Users/kunwoopark/workspace/fork/texera/amber/src/main/python/core/architecture/managers/state_manager.py", line 73, in transit_to
    raise InvalidTransitionException(
core.architecture.managers.state_manager.InvalidTransitionException: Cannot transit from READY to COMPLETED

[mengw15]

/take