Copilot commented on code in PR #8038:
URL: https://github.com/apache/texera/pull/8038#discussion_r3869469177
##########
amber/src/test/scala/org/apache/texera/web/ComputingUnitMasterSpec.scala:
##########
@@ -318,6 +357,171 @@ class ComputingUnitMasterSpec
}
}
+ /**
+ * The context every client below is built for. Deliberately NOT `new
WorkflowContext()`:
+ * the default ids are exactly what a factory that dropped its
`workflowContext` argument
+ * would fall back to, and a fixture that hands production the mutant's own
literal cannot
+ * tell the two apart.
+ */
+ private def specWorkflowContext: WorkflowContext =
+ new WorkflowContext(
+ workflowId = WorkflowIdentity(90210L),
+ executionId = ExecutionIdentity(90211L)
+ )
+
+ /**
+ * Builds a client the way `WorkflowExecutionService` does. The empty plan
plus an
+ * all-None config is the recipe AmberClientSpec and three `web.service`
specs already
+ * use: the AmberClient constructor blocks on an InitializeRequest that
spawns a real
+ * Coordinator child, and an empty plan lets that finish with no engine
behind it. The
+ * client is always shut down afterwards -- amber's suites share one
serially-run JVM, so
+ * a leaked client would leave a live actor tree behind for every later
suite.
+ */
+ private def withAmberRuntime(
+ workflowContext: WorkflowContext = specWorkflowContext,
+ physicalPlan: PhysicalPlan = PhysicalPlan(Set.empty, Set.empty),
+ conf: CoordinatorConfig = CoordinatorConfig(None, None, None, None),
+ errorHandler: Throwable => Unit = _ => ()
+ )(body: AmberClient => Unit): Unit = {
+ val client = ComputingUnitMaster.createAmberRuntime(
+ workflowContext,
+ physicalPlan,
+ conf,
+ errorHandler
+ )
+ try body(client)
+ finally client.shutdown()
+ }
+
+ /**
+ * Registers one SessionState against a stubbed websocket for the duration
of `body` and
+ * hands it the JSON payloads pushed to that session. `SessionState` keeps
a JVM-global
+ * registry, so the entry is always removed again -- a leaked one would
receive events
+ * from every later suite that builds a coordinator.
+ */
+ private def withRegisteredSession(body: (() => Seq[String]) => Unit): Unit =
{
+ val pushed = new ConcurrentLinkedQueue[String]()
+ val remote = stub(classOf[RemoteEndpoint.Async]) {
+ case ("sendText", args) => pushed.add(args.head.asInstanceOf[String]);
null
+ }
+ val session = stub(classOf[Session]) {
+ case ("getAsyncRemote", _) => remote.asInstanceOf[AnyRef]
+ }
+ val sessionId = "computing-unit-master-spec-" + java.util.UUID.randomUUID()
+ SessionState.setState(sessionId, new SessionState(session))
+ try body(() => pushed.asScala.toList)
+ finally SessionState.removeState(sessionId)
+ }
+
+ /** File names directly inside `folder`. */
+ private def entriesIn(folder: Path): Seq[String] = {
+ val listing = Files.list(folder)
+ try listing.iterator().asScala.map(_.getFileName.toString).toList
+ finally listing.close()
+ }
+
+ /**
+ * Reads the only field of `fieldType` off an AmberClient. Every field
there is
+ * class-private, and the client actor's is additionally name-mangled, so
they are looked
+ * up by TYPE rather than by a brittle spelling of the name.
+ */
+ private def amberClientField[T](client: AmberClient, fieldType: Class[T]): T
= {
+ val field = classOf[AmberClient].getDeclaredFields
+ .find(candidate => fieldType.isAssignableFrom(candidate.getType))
+ .getOrElse(fail(s"AmberClient no longer holds a
${fieldType.getSimpleName} field"))
+ field.setAccessible(true)
+ field.get(client).asInstanceOf[T]
+ }
+
+ "createAmberRuntime" should "build the client on the process-wide actor
system" in {
+ withAmberRuntime() { client =>
+ val clientActor = amberClientField(client, classOf[ActorRef])
+
+ // The factory has to hand AmberClient the process-wide
AmberRuntime.actorSystem and
+ // not one of its own: in production that is the system startActorMaster
bound to
+ // artery and joined to the cluster, and a workflow run on a private
system would be
+ // unreachable from the rest of the runtime. Identity is asserted as
well as the name
+ // -- resolving the path inside this suite's own system fails if the
actor was created
+ // anywhere else, even in a second system that happened to carry the
same name.
+ clientActor.path.address.system shouldBe testSystem.name
+ val resolved = Await.result(
+ testSystem
+ .actorSelection("/" + clientActor.path.elements.mkString("/"))
+ .resolveOne(10.seconds),
+ 10.seconds
+ )
+ resolved shouldBe clientActor
+ }
+ }
+
+ it should "hand the client the caller's error handler" in {
+ val errorHandler: Throwable => Unit = _ => ()
+
+ withAmberRuntime(errorHandler = errorHandler) { client =>
+ // AmberClient routes everything a registered callback throws to this
function and
+ // nowhere else, so a factory that quietly substituted a handler of its
own would
+ // swallow callback failures at the single place production builds a
client.
+ amberClientField(client, classOf[Function1[_, _]]) should be
theSameInstanceAs errorHandler
+ }
+ }
+
+ it should "forward the workflow context, physical plan and coordinator
config it is handed" in {
+ // The two assertions above pin arguments 1 and 5. Arguments 2, 3 and 4
are consumed
+ // inside the AmberClient constructor and retained on no field of it
(`javap -p` keeps
+ // only errorHandler, clientActor, timeout, registeredObservables,
isActive and
+ // coordinatorInterface), so they can only be observed through what the
coordinator the
+ // constructor spawns then does with them. Production passes
`workflow.context`,
+ // `workflow.physicalPlan` and the service's own config at
WorkflowExecutionService:124;
+ // a factory that substituted defaults for any of the three would run
every workflow with
+ // an empty plan, no workflow/execution identity and fault tolerance
disabled.
+ val scanOp = TestOperators.headerlessSmallCsvScanOpDesc()
+ val context = specWorkflowContext
+ val physicalPlan = buildWorkflow(List(scanOp), List.empty,
context).physicalPlan
+ val logFolder =
Files.createTempDirectory("computing-unit-master-spec-fault-tolerance")
+ val conf =
+ CoordinatorConfig(None, None, None, Some(FaultToleranceConfig(writeTo =
logFolder.toUri)))
+ // s"${toShorterString(COORDINATOR)}] [<simple name>" is AmberLogging's
naming scheme.
+ val scheduleGeneratorLogger =
+ s"${VirtualIdentityUtils.toShorterString(COORDINATOR)}]
[CostBasedScheduleGenerator"
+
Review Comment:
The logger name here hard-codes "CostBasedScheduleGenerator". This is easy
to drift from the actual class name used by AmberLogging (and won't fail fast
if it changes). Prefer deriving it from the class so renames either keep
working or fail at compile time.
--
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]
