aglinxinyuan commented on code in PR #8938:
URL: https://github.com/apache/texera/pull/8938#discussion_r4264506987
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala:
##########
@@ -116,20 +116,12 @@ class InputManager(
def getPort(portId: PortIdentity): WorkerPort = ports(portId)
/**
- * For ports that read from materialization, the port completion is marked
by the finish of the reader thread.
- * For other ports that connect to upstream links, the completion is marked
by the completion the port.
+ * Whether END_CHANNEL has been processed on every channel of the port
(`EndChannelHandler` then
+ * marks it completed), for a port read from materialization as well. Its
reader thread flags
+ * itself finished only after enqueuing END, so the DP thread can process
END before the flag is
+ * set, and the flag can be set while END is still queued.
*/
- def isPortCompleted(portId: PortIdentity): Boolean = {
- if (
- !this.inputPortMaterializationReaderThreads
- .contains(portId) ||
this.inputPortMaterializationReaderThreads(portId).isEmpty
- ) {
- this.getPort(portId).completed
- } else {
- val existingThread =
this.inputPortMaterializationReaderThreads(portId).head
- existingThread.finished
- }
- }
+ def isPortCompleted(portId: PortIdentity): Boolean =
this.getPort(portId).completed
Review Comment:
Good call — removed it in bb46d20ad7, along with its `AtomicBoolean`. The
tests that read it now check what it stood for: the END_CHANNEL the reader
enqueued, or the thread's own state (`NEW` / `TERMINATED`) in the two
port-completion tests. Also merged main.
--
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]
