aglinxinyuan commented on code in PR #6147:
URL: https://github.com/apache/texera/pull/6147#discussion_r3524542117
##########
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:
Good call — the test now proves the emit: it adds a downstream OneToOne
partitioner and captures the output handler, asserting a `StateFrame` is
actually emitted (110c93b).
##########
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:
Now proves no-emit: same downstream partitioner, but with `processState`
returning `None` it asserts NO `StateFrame` reaches the captured handler
(110c93b).
##########
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:
Now asserts the intended side effect: `dp.pauseManager.isPaused` is true
after `handleExecutorException` engages the operator-logic pause (110c93b).
##########
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:
Same — the tuple-exception test now asserts `dp.pauseManager.isPaused`
(routed through `handleExecutorException`) (110c93b).
##########
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:
Now asserts both effects: the operator is paused and the output iterator is
reset (`hasUnfinishedOutput` == false) (110c93b).
--
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]
