carloea2 commented on code in PR #8432:
URL: https://github.com/apache/texera/pull/8432#discussion_r4021215745
##########
amber/src/main/python/core/models/internal_marker.py:
##########
@@ -16,6 +16,9 @@
# under the License.
+from dataclasses import dataclass
+
+
class InternalMarker:
"""
A special Data Message, only being generated in un-packaging a batch into
Tuples.
Review Comment:
Updated the docstring to cover both batch unpacking and control handlers.
##########
amber/src/test/python/core/runnables/test_data_processor.py:
##########
@@ -71,19 +71,25 @@ def on_finish(self, port_id):
class TestProcessInternalMarker:
+ @pytest.mark.parametrize("marker", [StartChannel, EndChannel])
+ @pytest.mark.parametrize("port_id", [-1, True, "1"])
+ def test_channel_markers_reject_invalid_port_identity(self, marker,
port_id):
+ with pytest.raises(ValueError, match="nonnegative integer"):
+ marker(port_id)
+
@pytest.mark.timeout(2)
def test_start_channel_invokes_produce_state_on_start(
self, context, data_processor
):
executor = _StubExecutor()
context.executor_manager.executor = executor
- data_processor.process_internal_marker(StartChannel())
+ context.tuple_processing_manager.current_input_port_id = object()
+ data_processor.process_internal_marker(StartChannel(2))
Review Comment:
Added both handler-to-processor cases, including the port 0 open-state
assertion. Both fail with the old routing and pass with the fix. All 66 focused
tests pass.
##########
amber/src/main/python/core/runnables/data_processor.py:
##########
@@ -65,8 +65,8 @@ def run(self) -> None:
else:
self.process_tuple()
- def process_internal_marker(self, internal_marker: InternalMarker) -> None:
- with self._executor_session() as (executor, port_id):
+ def process_internal_marker(self, internal_marker: PortMarker) -> None:
+ with self._executor_session(internal_marker.port_id) as (executor,
port_id):
Review Comment:
Added the PortMarker guard with a TypeError before entering the executor,
plus regression tests for invalid markers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
