Xiao-zhen-Liu commented on code in PR #8938:
URL: https://github.com/apache/texera/pull/8938#discussion_r4263187206
##########
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:
Nit: after this change nothing in production reads the reader thread's
`finished` flag; only the new tests do. I'd remove it, so it can't be used for
completion again.
--
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]
