Copilot commented on code in PR #4490: URL: https://github.com/apache/texera/pull/4490#discussion_r3205800249
########## amber/src/test/python/core/storage/runnables/test_input_port_materialization_reader_runnable.py: ########## @@ -0,0 +1,99 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from unittest.mock import MagicMock, patch + +import pytest + +from core.models import State, StateFrame +from core.models.internal_queue import DataElement +from core.models.schema import Schema +from core.storage.runnables.input_port_materialization_reader_runnable import ( + InputPortMaterializationReaderRunnable, +) +from proto.org.apache.texera.amber.core import ( + ActorVirtualIdentity, + ChannelIdentity, +) + + +class TestRunStateReadingBlock: + """Cover the state-reading block in run() that opens the state + document and emits its rows as StateFrames directly to the input + queue (no partitioner filtering -- state is broadcast to every + worker). + """ + + @pytest.fixture + def me(self): + return ActorVirtualIdentity(name="me") + + @pytest.fixture + def runnable(self, me): + instance = InputPortMaterializationReaderRunnable.__new__( + InputPortMaterializationReaderRunnable + ) + instance.uri = "vfs:///wf/0/exec/0/result/op-a" + instance.worker_actor_id = me Review Comment: This test sets `instance.uri` to a non-base VFS path (it already contains `/result/...`), but the runnable now treats `uri` as a *port base URI* and derives both `result_uri(uri)` and `state_uri(uri)`. Using a realistic base URI here and asserting the exact URIs passed to `DocumentFactory.open_document` would make the test actually validate the new base-URI routing logic (and prevent future regressions). ########## amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/managers/InputPortMaterializationReaderThread.scala: ########## @@ -79,13 +84,34 @@ class InputPortMaterializationReaderThread( /** * Read from the materialization stoage, and mimcs the behavior of an upstream worker's output manager. Review Comment: Fix typos in the Scaladoc: “stoage” -> “storage” and “mimcs” -> “mimics”. ########## amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OutputManager.scala: ########## @@ -251,7 +264,11 @@ class OutputManager( writerThread.getFailure.foreach(throw _) case None => } - + this.stateWriterThreads.remove(outputPortId).foreach { writerThread => + writerThread.queue.put(Right(PortStorageWriterTerminateSignal)) + writerThread.join() + writerThread.getFailure.foreach(throw _) + } Review Comment: `stateWriterThreads` is removed from the map on close, but `outputPortResultWriterThreads` is not. This inconsistency means a closed result-writer thread remains addressable and can still receive enqueues (unbounded queue growth) if any late `saveTupleToStorageIfNeeded` calls occur. Consider also removing the result writer entry (e.g., `outputPortResultWriterThreads.remove(outputPortId)`) when closing, mirroring the state-writer cleanup. -- 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]
