aglinxinyuan commented on code in PR #6312:
URL: https://github.com/apache/texera/pull/6312#discussion_r3556447000
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DPThreadSpec.scala:
##########
@@ -270,4 +276,76 @@ class DPThreadSpec extends AnyFlatSpec with MockFactory {
assert(uri.length > baseUri.length)
}
+ "DP Thread" should "toggle backpressureStatus on Backpressure and ignore
other ActorCommands" in {
+ // Single-threaded: exercise handleActorCommand directly, without start(),
so the
+ // read of backpressureStatus is on the same thread that mutated it.
+ val q = new LinkedBlockingQueue[DPInputQueueElement]()
+ val dp = new DataProcessor(workerId, _ => {}, inputMessageQueue = q)
+ val dpThread = new DPThread(workerId, dp, logManager, q)
+
+ dpThread.handleActorCommand(Backpressure(enableBackpressure = true))
+ assert(dpThread.backpressureStatus)
+
+ dpThread.handleActorCommand(Backpressure(enableBackpressure = false))
+ assert(!dpThread.backpressureStatus)
+
+ // CreditUpdate is the other ActorCommand oneof subtype. It falls through
to the
+ // `case _ => // no op` arm and must leave backpressureStatus untouched.
+ dpThread.handleActorCommand(CreditUpdate())
+ assert(!dpThread.backpressureStatus)
+ }
+
+ "DP Thread" should "treat a second start() as a no-op" in {
+ val inputQueue = new LinkedBlockingQueue[DPInputQueueElement]()
+ val dp = new DataProcessor(workerId, x => {}, inputMessageQueue =
inputQueue)
+ dp.adaptiveBatchingMonitor = mock[WorkerTimerService]
+ (dp.adaptiveBatchingMonitor.resumeAdaptiveBatching
_).expects().anyNumberOfTimes()
+ val dpThread = new DPThread(workerId, dp, logManager, inputQueue)
+ try {
+ dpThread.start()
+ val executorAfterFirstStart = dpThread.dpThreadExecutor
+ val futureAfterFirstStart = dpThread.dpThread
+ // The second start() should log "already running" and change nothing:
+ // no new executor, no second worker thread.
+ dpThread.start()
+ assert(dpThread.dpThreadExecutor eq executorAfterFirstStart)
+ assert(dpThread.dpThread eq futureAfterFirstStart)
+ } finally {
+ dpThread.stop()
+ }
+ }
+
+ "DP Thread" should "forward an uncaught processing error to the main thread"
in {
+ // A processing error escaping the main logic must be caught in run() and
delegated
+ // back to the main thread as a Left(MainThreadDelegateMessage). We throw
from
+ // DataProcessor.processDataPayload (not from a mock executor:
DataProcessor swallows
+ // executor exceptions in handleExecutorException, so they never reach
DPThread).
+ val inputQueue = new LinkedBlockingQueue[DPInputQueueElement]()
+ val captured =
+ new CompletableFuture[Either[MainThreadDelegateMessage,
WorkflowFIFOMessage]]()
+ val outputHandler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage]
=> Unit =
+ e => captured.complete(e)
+ val dp = new DataProcessor(workerId, outputHandler, inputMessageQueue =
inputQueue) {
+ override def processDataPayload(
+ channelId: ChannelIdentity,
+ dataPayload: DataPayload
+ ): Unit = throw new RuntimeException("boom")
+ }
+ dp.inputManager.addPort(mockInputPortId, schema, List.empty, List.empty)
+ dp.inputGateway.getChannel(dataChannelId).setPortId(mockInputPortId)
+ dp.adaptiveBatchingMonitor = mock[WorkerTimerService]
+ (dp.adaptiveBatchingMonitor.resumeAdaptiveBatching
_).expects().anyNumberOfTimes()
+ val dpThread = new DPThread(workerId, dp, logManager, inputQueue)
+ try {
+ dpThread.start()
+ inputQueue.put(
+ FIFOMessageElement(WorkflowFIFOMessage(dataChannelId, 0,
DataFrame(Array(tuples(0)))))
+ )
+ val result = captured.get(5, TimeUnit.SECONDS)
+ assert(result.isLeft)
+ } finally {
Review Comment:
Strengthened it — the test now pattern-matches the
`Left(MainThreadDelegateMessage(closure))`, runs `closure(null)`, and asserts
it re-throws the original `RuntimeException("boom")` (43443d5).
--
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]
