Copilot commented on code in PR #6147:
URL: https://github.com/apache/texera/pull/6147#discussion_r3524524957
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorSpec.scala:
##########
@@ -241,4 +247,138 @@ class DataProcessorSpec extends AnyFlatSpec with
MockFactory with BeforeAndAfter
}
}
+ "data processor" should "process a state frame and emit the produced state"
in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 1))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .returning(Some(inputState))
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
Review Comment:
These StateFrame tests currently don't actually validate the emit/no-emit
behavior: `OutputManager.emitState` only sends when
`addPartitionerWithPartitioning` has populated `networkOutputBuffers`, and the
spec doesn't set any partitioner. As a result, even if `emitState` were
mistakenly skipped (or always called), these tests would still pass. Set up a
simple downstream partitioner and assert the expected `StateFrame` is (or is
not) sent via `outputHandler`.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorSpec.scala:
##########
@@ -241,4 +247,138 @@ class DataProcessorSpec extends AnyFlatSpec with
MockFactory with BeforeAndAfter
}
}
+ "data processor" should "process a state frame and emit the produced state"
in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 1))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .returning(Some(inputState))
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ StateFrame(inputState)
+ )
+ }
+ }
+
+ "data processor" should "not emit when processState yields None" in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 2))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .returning(None)
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ StateFrame(inputState)
+ )
+ }
+ }
+
+ "data processor" should "handle an exception thrown while processing a state
frame" in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 3))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .throwing(new RuntimeException("boom on state"))
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ StateFrame(inputState)
+ )
+ }
+ }
+
+ "data processor" should "handle an exception thrown while processing an
input tuple" in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ (
+ (
+ tuple: Tuple,
+ input: Int
+ ) => executor.processTupleMultiPort(tuple, input)
+ )
+ .expects(tuples.head, 0)
+ .throwing(new RuntimeException("boom on tuple"))
+ (adaptiveBatchingMonitor.startAdaptiveBatching
_).expects().anyNumberOfTimes()
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ DataFrame(Array(tuples.head))
+ )
+ }
+ }
Review Comment:
This tuple exception test should also assert that the executor exception is
routed through `handleExecutorException` (pause is engaged), not just that the
exception is swallowed. That makes the test robust against regressions where
the exception is caught but no pause is triggered.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorSpec.scala:
##########
@@ -241,4 +247,138 @@ class DataProcessorSpec extends AnyFlatSpec with
MockFactory with BeforeAndAfter
}
}
+ "data processor" should "process a state frame and emit the produced state"
in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 1))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .returning(Some(inputState))
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ StateFrame(inputState)
+ )
+ }
+ }
+
+ "data processor" should "not emit when processState yields None" in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 2))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .returning(None)
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
Review Comment:
This "processState yields None" test doesn't currently prove the no-emit
behavior because there are no downstream partitioners, so `emitState` would
produce no observable output even if it were called. Configure a downstream
partitioner and make the expectation strict (no `outputHandler` calls) so the
test fails if a `StateFrame` is emitted.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorSpec.scala:
##########
@@ -241,4 +247,138 @@ class DataProcessorSpec extends AnyFlatSpec with
MockFactory with BeforeAndAfter
}
}
+ "data processor" should "process a state frame and emit the produced state"
in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 1))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .returning(Some(inputState))
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ StateFrame(inputState)
+ )
+ }
+ }
+
+ "data processor" should "not emit when processState yields None" in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 2))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .returning(None)
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ StateFrame(inputState)
+ )
+ }
+ }
+
+ "data processor" should "handle an exception thrown while processing a state
frame" in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 3))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .throwing(new RuntimeException("boom on state"))
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ StateFrame(inputState)
+ )
+ }
+ }
+
+ "data processor" should "handle an exception thrown while processing an
input tuple" in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ (
+ (
+ tuple: Tuple,
+ input: Int
+ ) => executor.processTupleMultiPort(tuple, input)
+ )
+ .expects(tuples.head, 0)
+ .throwing(new RuntimeException("boom on tuple"))
+ (adaptiveBatchingMonitor.startAdaptiveBatching
_).expects().anyNumberOfTimes()
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ DataFrame(Array(tuples.head))
+ )
+ }
+ }
+
+ "data processor" should "handle an exception thrown while advancing the
output iterator" in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ (adaptiveBatchingMonitor.startAdaptiveBatching
_).expects().anyNumberOfTimes()
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ // Poison the output iterator: hasNext is true so continueDataProcessing
routes
+ // into outputOneTuple, but next() throws to exercise the catch branch.
+ dp.outputManager.outputIterator.setTupleOutput(
+ new Iterator[(TupleLike, Option[PortIdentity])] {
+ override def hasNext: Boolean = true
+ override def next(): (TupleLike, Option[PortIdentity]) =
+ throw new RuntimeException("boom on next")
+ }
+ )
+ assert(dp.outputManager.hasUnfinishedOutput)
+ noException should be thrownBy {
+ dp.continueDataProcessing()
+ }
+ }
Review Comment:
For the poisoned output iterator case, consider asserting both that the
operator is paused (via `handleExecutorException`) and that the output iterator
was invalidated (so `hasUnfinishedOutput` becomes false). Otherwise the test
only checks non-propagation and could miss regressions in the recovery/cleanup
logic inside `outputOneTuple`.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorSpec.scala:
##########
@@ -241,4 +247,138 @@ class DataProcessorSpec extends AnyFlatSpec with
MockFactory with BeforeAndAfter
}
}
+ "data processor" should "process a state frame and emit the produced state"
in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 1))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .returning(Some(inputState))
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ StateFrame(inputState)
+ )
+ }
+ }
+
+ "data processor" should "not emit when processState yields None" in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 2))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .returning(None)
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ StateFrame(inputState)
+ )
+ }
+ }
+
+ "data processor" should "handle an exception thrown while processing a state
frame" in {
+ val dp = mkDataProcessor
+ dp.executor = executor
+ dp.stateManager.transitTo(READY)
+ (outputHandler.apply _).expects(*).anyNumberOfTimes()
+ val inputState = State(Map("field1" -> 3))
+ (
+ (
+ state: State,
+ port: Int
+ ) => executor.processState(state, port)
+ )
+ .expects(inputState, 0)
+ .throwing(new RuntimeException("boom on state"))
+ dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty)
+ dp.inputGateway
+ .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl =
false))
+ .setPortId(inputPortId)
+ dp.outputManager.addPort(outputPortId, schema, None)
+ noException should be thrownBy {
+ dp.processDataPayload(
+ ChannelIdentity(senderWorkerId, testWorkerId, isControl = false),
+ StateFrame(inputState)
+ )
+ }
+ }
Review Comment:
These exception-path tests only assert that exceptions don't propagate, but
they don't assert the intended side effect of `handleExecutorException`
(pausing operator logic). Adding an explicit check on
`dp.pauseManager.isPaused` makes the test actually validate the behavior
described in the PR.
--
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]
