sshiv012 commented on code in PR #8432:
URL: https://github.com/apache/texera/pull/8432#discussion_r4019623127
##########
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:
Minor, non-blocking: could we make the InternalMarker docstring more
generic? “Only being generated in un-packaging a batch” doesn’t cover markers
created by 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:
Can we add a complete two port regression test to improve coverage?
Especially to test the integration of handler and data processor. Eg. Register
input ports 0 and 1, process a data tuple on port 0, deliver end channel for
empty port 1, assert produce_state_on_finish and on_finish both receive port 1.
Assert port 0 remains open. Another case could deliver StartChannel on a
nonzero port before any data arrives and verify that produce_state_on_start
receives that port.
##########
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:
An isinstance(internal_marker, PortMarker) guard with an else: raise.. would
be a neat check to do here?
--
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]
