From f473f28218634f01e07994a2bc9910ad48ee0ebb Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Fri, 14 Aug 2026 02:34:58 -0700 Subject: [PATCH] test(amber): cover the runtime's serde bootstrap and worker startup AmberRuntime had no spec. It owns the actor system the rest of amber leans on, the serialization extension every message goes through, and the scheduling helpers services use for recurring work - and none of it was verified. Adds 15 tests, taking the file from 12.20% to 65.85% of lines. Nothing needs a cluster: startActorWorker(None) builds its whole system with artery on port 0 in under 100ms, makes no network call, and tears down again, so the worker startup path is drivable end to end. Covered: the serde bootstrap and its memoization, extending an installed system rather than replacing it, both scheduling helpers including cancellation and the recurring delay, the seed-node URI and master address, the artery host binding, the config fallback chain, and the two child actors with their dead-letter subscription. The spec saves and restores AmberRuntime's globals by reflection, the same way eight sibling specs do, and terminates every system it starts; run alongside ClusterListenerSpec and SessionStateSpec, all 21 pass. Left uncovered: getNodeIpAddress and startActorMaster, which bind real network interfaces. No production file is touched. --- .../engine/common/AmberRuntimeSpec.scala | 404 ++++++++++++++++++ 1 file changed, 404 insertions(+) create mode 100644 amber/src/test/scala/org/apache/texera/amber/engine/common/AmberRuntimeSpec.scala diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/common/AmberRuntimeSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/common/AmberRuntimeSpec.scala new file mode 100644 index 00000000000..510764689db --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/amber/engine/common/AmberRuntimeSpec.scala @@ -0,0 +1,404 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.engine.common + +import com.typesafe.config.Config +import org.apache.pekko.actor.{ActorRef, ActorSystem, Address, DeadLetter} +import org.apache.pekko.serialization.{Serialization, SerializationExtension} +import org.apache.pekko.testkit.TestKit +import org.apache.texera.amber.clustering.ClusterListener +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpec + +import java.util.concurrent.CountDownLatch +import java.util.concurrent.atomic.AtomicInteger +import scala.concurrent.Await +import scala.concurrent.duration._ +import scala.jdk.CollectionConverters.ListHasAsScala +import scala.util.Try + +/** + * Characterizes [[AmberRuntime]] — the process-wide holder of the engine's Pekko + * `ActorSystem` and `Serialization`, and the code that builds them. + * + * What this spec catches: + * - `serde` losing its memoization, or bootstrapping a system when one is already + * installed (either would hand different halves of the engine different + * serializers, and the bootstrap branch leaks an untracked cluster system), + * - the schedule helpers ignoring their delays, or handing back a `Cancellable` + * that does not actually cancel (the reconfiguration and monitoring paths rely + * on both), + * - a regression in `startActorWorker`'s config assembly: this is the entire + * startup path of `ComputingUnitWorker`, and the seed-node URI, the artery + * hostname and the `withFallback(pekkoConfig)` are what make a worker join the + * right cluster at all, + * - `createAmberSystem` renaming/dropping the `cluster-info` or + * `dead-letter-monitor-actor` children, or dropping the `DeadLetter` + * subscription — that last one silently disables dead-letter forwarding, with + * nothing failing anywhere. + * + * Deliberately NOT covered: + * - `getNodeIpAddress` opens `http://checkip.amazonaws.com` with no parameter, + * no overridable member and no config knob to redirect it, so neither its + * success nor its (no-op rethrow) failure arm is reachable from a test. That + * also makes the `mainNodeAddress.isDefined` arm of `startActorWorker` + * untestable, hence `startActorWorker(None)` throughout. + * - `startActorMaster` binds artery on the FIXED port 2552 and lists itself as a + * seed node, so the node self-joins and reaches Up. That fires a real + * `MemberUp` into `ClusterListener.updateClusterStatus`, which fans a + * `ClusterStatusUpdateEvent` out to every entry of the JVM-global + * `SessionState` registry — and amber runs its suites in ONE shared JVM, where + * `ClusterListenerSpec` holds live per-test ScalaMock sessions in exactly that + * registry. `ComputingUnitMasterSpec` refuses this path for the same reason. + * `createMasterAddress` is still covered, via `startActorWorker`. + * + * JVM-global state: this object IS the global other amber suites depend on, and sbt + * runs them concurrently in one JVM without forking. Every global this spec touches + * (`_actorSystem`, `_serde`, `AmberConfig.masterNodeAddr`, + * `ClusterListener.numWorkerNodesInCluster`) is saved in `beforeAll` and put back in + * `afterAll` — the save/restore camp of `HDFSRecordStorageSpec` / + * `ComputingUnitMasterSpec`, not the null-out camp, because nulling makes + * `AmberRuntime.serde` take its bootstrap branch mid-run inside a sibling suite and + * spin up a system nobody shuts down. Tests that have to write those globals + * themselves restore them in a `finally` rather than deferring to `afterAll`, so the + * write window is one test long. + */ +class AmberRuntimeSpec extends AnyFlatSpec with BeforeAndAfterAll { + + // `lazy` matters: BeforeAndAfterAll skips beforeAll/afterAll entirely when the + // active filter selects no test from this suite (which is what the + // amber-integration CI job does via AMBER_TEST_FILTER). A strict `val` would + // still build an ActorSystem during suite construction and never shut it down. + private lazy val testSystem: ActorSystem = + ActorSystem("AmberRuntimeSpec-test", AmberRuntime.pekkoConfig) + private lazy val testSerde: Serialization = SerializationExtension(testSystem) + + private def getAmberRuntimeField(name: String): AnyRef = { + val field = AmberRuntime.getClass.getDeclaredField(name) + field.setAccessible(true) + field.get(AmberRuntime) + } + + private def setAmberRuntimeField(name: String, value: AnyRef): Unit = { + val field = AmberRuntime.getClass.getDeclaredField(name) + field.setAccessible(true) + field.set(AmberRuntime, value) + } + + private var previousActorSystem: AnyRef = _ + private var previousSerde: AnyRef = _ + private var previousMasterNodeAddr: Address = _ + private var previousNodeCount: Int = 0 + + override protected def beforeAll(): Unit = { + super.beforeAll() + previousActorSystem = getAmberRuntimeField("_actorSystem") + previousSerde = getAmberRuntimeField("_serde") + previousMasterNodeAddr = AmberConfig.masterNodeAddr + previousNodeCount = ClusterListener.numWorkerNodesInCluster + setAmberRuntimeField("_actorSystem", testSystem) + setAmberRuntimeField("_serde", testSerde) + } + + override protected def afterAll(): Unit = { + setAmberRuntimeField("_serde", previousSerde) + setAmberRuntimeField("_actorSystem", previousActorSystem) + AmberConfig.masterNodeAddr = previousMasterNodeAddr + // Belt and braces: the worker system's seed list points at localhost:2552, so on + // a host that happens to be running a local coordinator it could in principle + // join and have ClusterListener bump this count. + ClusterListener.numWorkerNodesInCluster = previousNodeCount + TestKit.shutdownActorSystem(testSystem) + super.afterAll() + } + + /** + * Installs `actorSystem` / `serde` into the two private statics for the duration of + * `body`, then puts the suite's own pair back. + * + * Restoring inside the test rather than in `afterAll` keeps the window in which a + * sibling suite could observe a null (and take the leaking bootstrap branch) to the + * length of one test. + */ + private def withGlobals(actorSystem: AnyRef, serde: AnyRef)(body: => Unit): Unit = { + setAmberRuntimeField("_actorSystem", actorSystem) + setAmberRuntimeField("_serde", serde) + try body + finally { + setAmberRuntimeField("_actorSystem", testSystem) + setAmberRuntimeField("_serde", testSerde) + } + } + + // --------------------------------------------------------------------------- + // serde + // --------------------------------------------------------------------------- + + "AmberRuntime.serde" should "bootstrap its own Amber system when none is installed" in { + // The bootstrap branch is what any engine component that serializes before + // startActorWorker/startActorMaster has run ends up on. + withGlobals(null, null) { + val serde = AmberRuntime.serde + assert(serde != null) + assert(serde.system.name == "Amber") + // The bootstrapped system is assigned to _serde ONLY - _actorSystem stays + // null, so AmberRuntime.actorSystem keeps returning null afterwards. Pinning + // the quirk because it is the reason this branch leaks a system unless the + // caller reaches it back through `serde.system`, which is what we do below. + assert(getAmberRuntimeField("_actorSystem") == null) + TestKit.shutdownActorSystem(serde.system) + } + } + + it should "extend the installed actor system when one is present" in { + withGlobals(testSystem, null) { + // `eq`, not `==`: the contract is that the serde is built over the SAME system + // the rest of the engine uses. A bootstrap here would produce a serde that + // cannot deserialize actor refs from the installed system. + assert(AmberRuntime.serde.system eq testSystem) + } + } + + it should "memoize, returning the installed Serialization untouched" in { + // _actorSystem is null on purpose: with the memoization guard dropped, the call + // falls into the bootstrap branch and returns a Serialization over a brand-new + // system, so the identity check below is what fails. + withGlobals(null, testSerde) { + assert(AmberRuntime.serde eq testSerde) + assert(getAmberRuntimeField("_serde") eq testSerde) + } + } + + // --------------------------------------------------------------------------- + // actorSystem + // --------------------------------------------------------------------------- + + "AmberRuntime.actorSystem" should "hand back the installed system" in { + withGlobals(testSystem, testSerde) { + assert(AmberRuntime.actorSystem eq testSystem) + } + } + + it should "be null before a system has been installed" in { + // Not redundant with the case above: callers such as ComputingUnitMaster read + // this before startup to decide whether the runtime is up at all. It is also the + // only assertion here that a serde-derived implementation could not satisfy - + // _serde is populated below, so `_serde.system` would answer testSystem. + withGlobals(null, testSerde) { + assert(AmberRuntime.actorSystem == null) + } + } + + // --------------------------------------------------------------------------- + // scheduleCallThroughActorSystem + // --------------------------------------------------------------------------- + + "AmberRuntime.scheduleCallThroughActorSystem" should "run the call once the delay elapses" in { + val latch = new CountDownLatch(1) + val cancellable = AmberRuntime.scheduleCallThroughActorSystem(100.millis)(latch.countDown()) + try assert(latch.await(5, java.util.concurrent.TimeUnit.SECONDS), "scheduled call never ran") + finally cancellable.cancel() + } + + it should "wait out the delay and stop firing once cancelled" in { + // The case above alone is a one-sided pin: it stays green if the delay is + // ignored, and green if the returned Cancellable is inert. Both halves are + // pinned here, and the second window deliberately runs PAST the 1s delay so an + // inert Cancellable would let the call through. + val fired = new AtomicInteger(0) + val cancellable = AmberRuntime.scheduleCallThroughActorSystem(1.second)(fired.incrementAndGet()) + try { + Thread.sleep(300) + assert(fired.get() == 0, "the call ran before its delay elapsed") + cancellable.cancel() + Thread.sleep(1200) + assert(fired.get() == 0, "the call ran after its Cancellable was cancelled") + } finally cancellable.cancel() + } + + // --------------------------------------------------------------------------- + // scheduleRecurringCallThroughActorSystem + // --------------------------------------------------------------------------- + + "AmberRuntime.scheduleRecurringCallThroughActorSystem" should + "honor the initial delay and then repeat at the fixed delay" in { + // The two durations are deliberately UNEQUAL, so they are pinned separately: + // with the arguments swapped (100ms initial, 1s repeat) the first assertion + // sees a fire inside 400ms, and with no repeat at all the count stops at 1. + val fired = new AtomicInteger(0) + val cancellable = AmberRuntime.scheduleRecurringCallThroughActorSystem( + initialDelay = 1.second, + delay = 100.millis + )(fired.incrementAndGet()) + try { + Thread.sleep(400) + assert(fired.get() == 0, "the recurring call ran before its initial delay elapsed") + val deadline = System.nanoTime() + 4.seconds.toNanos + while (fired.get() < 3 && System.nanoTime() < deadline) Thread.sleep(50) + assert(fired.get() >= 3, s"expected repeated fires, saw ${fired.get()}") + } finally cancellable.cancel() + } + + // --------------------------------------------------------------------------- + // startActorWorker(None) -> createAmberSystem + // --------------------------------------------------------------------------- + + /** + * Everything one `startActorWorker(None)` invocation makes observable. + * + * The whole startup costs about 175ms, so it is paid once and the observations + * are captured while the created system is alive; the system is shut down inside + * this block so its cluster membership window stays as short as production's own + * startup, and the globals it overwrites are restored here rather than in + * `afterAll`. + * + * The two child lookups are captured as `Option`s rather than asserted here so a + * rename fails only the case that is about that name. + */ + private case class WorkerStartup( + systemName: String, + config: Config, + masterNodeAddr: Address, + rawSerdeAfterCall: AnyRef, + serdeSystemIsCreatedSystem: Boolean, + clusterInfoPath: Option[String], + deadLetterMonitorPath: Option[String], + deadLetterResubscribeWasNew: Option[Boolean] + ) + + private lazy val workerStartup: WorkerStartup = { + // A sentinel is mandatory: AmberConfig.masterNodeAddr's DEFAULT is already + // Address("pekko", "Amber", "localhost", 2552), so without this the + // masterNodeAddr case would assert a value that held before the call. + AmberConfig.masterNodeAddr = Address("pekko", "Sentinel", "nowhere", 1) + setAmberRuntimeField("_actorSystem", null) + setAmberRuntimeField("_serde", null) + var created: ActorSystem = null + try { + AmberRuntime.startActorWorker(None) + // Read the raw field, not the AmberRuntime.actorSystem getter, so this fixture + // does not depend on the accessor that other cases here are pinning. + created = getAmberRuntimeField("_actorSystem").asInstanceOf[ActorSystem] + WorkerStartup( + systemName = created.name, + config = created.settings.config, + masterNodeAddr = AmberConfig.masterNodeAddr, + // Captured BEFORE anything calls AmberRuntime.serde: with _actorSystem now + // set, the getter would lazily rebuild an equivalent serde over the same + // system, so asserting through it would pass even with the write deleted. + rawSerdeAfterCall = getAmberRuntimeField("_serde"), + // Identity, not the system's name: a serde built over a different system that happens to + // be called "Amber" would satisfy a name comparison. + serdeSystemIsCreatedSystem = getAmberRuntimeField("_serde") match { + case serde: Serialization => serde.system eq created + case _ => false + }, + clusterInfoPath = resolveChild(created, "/user/cluster-info").map(_.path.toString), + deadLetterMonitorPath = + resolveChild(created, "/user/dead-letter-monitor-actor").map(_.path.toString), + deadLetterResubscribeWasNew = resolveChild(created, "/user/dead-letter-monitor-actor") + .map(ref => created.eventStream.subscribe(ref, classOf[DeadLetter])) + ) + } finally { + setAmberRuntimeField("_actorSystem", testSystem) + setAmberRuntimeField("_serde", testSerde) + if (created != null) TestKit.shutdownActorSystem(created) + } + } + + private def resolveChild(system: ActorSystem, path: String): Option[ActorRef] = + Try(Await.result(system.actorSelection(path).resolveOne(5.seconds), 5.seconds)).toOption + + "AmberRuntime.startActorWorker" should "bind artery on the local host" in { + // cluster.conf's own default hostname is "0.0.0.0", so this catches both a dropped + // hostname override and one that reaches for getNodeIpAddress on the None arm. + // The sibling `canonical.port = 0` override is deliberately NOT asserted: cluster.conf + // already defaults that key to 0, so removing the override changes no observable + // value and any assertion on it would be unfalsifiable (verified by mutation). + assert(workerStartup.config.getString("pekko.remote.artery.canonical.hostname") == "localhost") + } + + it should "seed the cluster from the master node's URI" in { + // cluster.conf ships `seed-nodes = []`, so the whole list is this method's work. + // The host segment comes from `mainNodeAddress.getOrElse("localhost")`, which is + // the only place the None arm's default is observable. + assert( + workerStartup.config + .getStringList("pekko.cluster.seed-nodes") + .asScala + .toList == List("pekko://Amber@localhost:2552") + ) + } + + it should "keep the rest of pekkoConfig underneath the overrides" in { + // The parsed override block mentions neither of these; both come from + // cluster.conf through withFallback. Without the fallback the worker would + // silently start on the local (non-cluster) provider and never join anything. + assert(workerStartup.config.getString("pekko.actor.provider") == "cluster") + // The kryo serializer, not artery's transport: `transport = tcp` is pekko's own + // reference.conf default, so asserting it holds even with pekkoConfig dropped from the chain. + // This binding exists only in cluster.conf. + assert( + workerStartup.config.getString("pekko.actor.serializers.kryo") == + "io.altoo.serialization.kryo.pekko.PekkoKryoSerializer" + ) + } + + it should "publish the master node address" in { + // Characterization, not a live contract: `AmberConfig.masterNodeAddr` is written + // here and by startActorMaster and read nowhere in the repo. It is asserted + // because it is the only observable output of `createMasterAddress`, so the + // address this method derives from `mainNodeAddress` stays pinned in case a + // reader comes back. + assert(workerStartup.masterNodeAddr == Address("pekko", "Amber", "localhost", 2552)) + } + + "AmberRuntime.createAmberSystem" should "name the system Amber and attach both children" in { + // Paths, which is what callers depend on: DeployStrategies and the frontend's cluster badge + // reach these actors by path, so a rename breaks them with no compile error. + // + // What this does NOT pin: which class is mounted at each path. Swapping ClusterListener for + // DeadLetterMonitorActor while keeping the name leaves this green, because an ActorRef exposes + // no class and nothing here exchanges a message only one of them answers. Recorded rather than + // chased -- pinning it would mean giving the production code a seam it does not otherwise need. + assert(workerStartup.systemName == "Amber") + assert(workerStartup.clusterInfoPath.contains("pekko://Amber/user/cluster-info")) + assert( + workerStartup.deadLetterMonitorPath.contains("pekko://Amber/user/dead-letter-monitor-actor") + ) + } + + it should "subscribe the dead-letter monitor to DeadLetter" in { + // `subscribe` answers false when the ref is already subscribed, so re-subscribing + // is both a no-op and the only observable proof the production call happened. + assert( + workerStartup.deadLetterResubscribeWasNew.contains(false), + "dead-letter monitor was not already subscribed to DeadLetter" + ) + } + + it should "install a serde over the system it created" in { + assert(workerStartup.rawSerdeAfterCall != null) + assert( + workerStartup.serdeSystemIsCreatedSystem, + "the installed serde is bound to the very system startActorWorker created" + ) + } +}