This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7544-ed16a605dacc0185bc5f069d14157ee3e128b68a in repository https://gitbox.apache.org/repos/asf/texera.git
commit 686aedba6c41a95a509f44f88e0c336456aac382 Author: Xinyuan Lin <[email protected]> AuthorDate: Tue Aug 11 00:14:53 2026 -0700 test(amber): cover the execution statistics service (#7544) ### What changes were proposed in this PR? `ExecutionStatsService` had no spec and sat at **0% of its 80 lines**, while being the class that tells the frontend what an execution is doing: per-operator input/output metrics, worker assignment, elapsed duration, and the runtime statistics persisted for the dashboard's time series. Adds 12 tests driven through the three-argument constructor. The client is an `AmberClient` subclass that captures the registered callbacks, so the tests fire `ExecutionStatsUpdate`, `RuntimeStatisticsPersist`, `WorkerAssignmentUpdate`, `WorkflowRecoveryStatus`, `FatalError` and `ExecutionStateUpdate` directly; the state store is real, so events travel through the production diff handlers. This follows `ExecutionRuntimeServiceSpec` and `ExecutionConsoleServiceSpec`. The most valuable one is the carry-forward: an operator that stops reporting must still appear in the persisted statistics, or its row silently vanishes from the time series mid-execution. ### Verification 19 mutations applied and reverted, production diff empty after each. All red, including the positional column layout of the persisted tuple, the commit guard holding statistics back until a terminal state, `client.shutdown()` on `FatalError`, and the wid/eid argument order into `updateRuntimeStatsUri`. Three assertions were **found to be vacuous in review and strengthened**, which is the part worth reading: | Weakness | Why it passed | Fix | |---|---|---| | running-duration arithmetic unpinned | the test asserted only `duration >= 1500`, so `currentTime - start` becoming `currentTime + start` (~111 years) still passed | bounded on both sides with a window captured around the update | | "every operator" observed one operator | the fixture reported a single operator, so `operatorInfo.collect` could be narrowed to `.take(1).collect` | a second operator with different numbers | | "publish nothing" checked the payload, not the event | the helper flattened the event's map, so a present-but-empty churn event to the websocket was invisible | collect by event type and assert empty, matching the two sibling tests | All three mutations are now red. ### Deliberately not included - The `catch` around `runtimeStatsWriter.close()` — Iceberg's close is idempotent, so nothing reaches it without injecting a throwing writer. - The `catch` in `storeRuntimeStatistics` — it runs on a private single-thread executor, which swallows the throwable, so no assertion could observe the mutation. - Three dead lines in `computeStatsDiff` (`defaultMetrics`, `newKeys`, and the `++ newKeys.map(...)` merge). `updatedLastMetrics` is read only at `oldKeys.map(key => key -> updatedLastMetrics(key))`, and `oldKeys`/`newKeys` are disjoint by construction, so the merged entries can never be selected — confirmed by replacing the whole expression with `lastPersistedMetrics` and seeing all 12 tests stay green. Reported rather than cemented; deleting beats testing. Two notes for reviewers. The spec uses distinct workflow/execution ids because the runtime-statistics URI derives from them and `createDocument` overrides an existing table — a default `WorkflowContext` collides with `DefaultCostEstimatorSpec`, and sbt runs amber suites in parallel in one JVM. And no temp Iceberg catalog is installed: `IcebergCatalogInstance.replaceInstance` is JVM-wide and the URI carries no warehouse, so installing one would hijack the catalog for every other amber suite. No production file is touched. ### Any related issues, documentation, discussions? Closes #7542 ### How was this PR tested? ``` STORAGE_ICEBERG_CATALOG_TYPE=postgres sbt "WorkflowExecutionService/testOnly org.apache.texera.web.service.ExecutionStatsServiceSpec" ``` ``` [info] Total number of tests run: 12 [info] Tests: succeeded 12, failed 0, canceled 0, ignored 0, pending 0 ``` The env var matches what CI's unit job already sets (`.github/workflows/build.yml:293`); the committed default expects a Lakekeeper. `Test/scalafmtCheck` and `Test/scalafix --check` both pass. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --- .../web/service/ExecutionStatsServiceSpec.scala | 585 +++++++++++++++++++++ 1 file changed, 585 insertions(+) diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionStatsServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionStatsServiceSpec.scala new file mode 100644 index 0000000000..acc34e0e7c --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionStatsServiceSpec.scala @@ -0,0 +1,585 @@ +/* + * 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.web.service + +import io.reactivex.rxjava3.disposables.Disposable +import org.apache.pekko.actor.ActorSystem +import org.apache.pekko.testkit.TestKit +import org.apache.texera.amber.core.storage.model.VirtualDocument +import org.apache.texera.amber.core.storage.{DocumentFactory, VFSURIFactory} +import org.apache.texera.amber.core.tuple.Tuple +import org.apache.texera.amber.core.virtualidentity.{ + ActorVirtualIdentity, + ExecutionIdentity, + WorkflowIdentity +} +import org.apache.texera.amber.core.workflow.{PhysicalPlan, PortIdentity, WorkflowContext} +import org.apache.texera.amber.core.workflowruntimestate.FatalErrorType.EXECUTION_FAILURE +import org.apache.texera.amber.engine.architecture.coordinator.{ + CoordinatorConfig, + ExecutionStateUpdate, + ExecutionStatsUpdate, + FatalError, + RuntimeStatisticsPersist, + WorkerAssignmentUpdate, + WorkflowRecoveryStatus +} +import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState +import org.apache.texera.amber.engine.architecture.rpc.controlreturns.WorkflowAggregatedState._ +import org.apache.texera.amber.engine.architecture.worker.statistics.{ + PortTupleMetricsMapping, + TupleMetrics +} +import org.apache.texera.amber.engine.common.client.AmberClient +import org.apache.texera.amber.engine.common.executionruntimestate.{ + ExecutionStatsStore, + OperatorMetrics, + OperatorStatistics +} +import org.apache.texera.dao.MockTexeraDB +import org.apache.texera.dao.jooq.generated.Tables.WORKFLOW_EXECUTIONS +import org.apache.texera.dao.jooq.generated.tables.daos.{ + UserDao, + WorkflowDao, + WorkflowExecutionsDao, + WorkflowVersionDao +} +import org.apache.texera.dao.jooq.generated.tables.pojos.{ + User, + Workflow, + WorkflowExecutions, + WorkflowVersion +} +import org.apache.texera.web.model.websocket.event.{ + ExecutionDurationUpdateEvent, + OperatorAggregatedMetrics, + OperatorStatisticsUpdateEvent, + TexeraWebSocketEvent, + WorkerAssignmentUpdateEvent +} +import org.apache.texera.web.storage.{ExecutionStateStore, StateStore} +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpecLike +import org.scalatest.matchers.should.Matchers + +import java.net.URI +import java.sql.Timestamp +import java.util.concurrent.ExecutorService +import scala.collection.mutable +import scala.collection.mutable.ListBuffer +import scala.reflect.ClassTag + +/** + * The service class was entirely uncovered. It owns two independent outputs: the websocket + * diff handlers that tell the frontend what each operator is doing, and the runtime-statistics + * iceberg table that the dashboard replays after the execution is over. + * + * Everything here runs against an empty-plan AmberClient, so no engine is involved: the test + * fires the client events straight into the callbacks the service registered. + * + * Two things about the ids below are deliberate. They must not collide with the + * runtime-statistics URI of any other suite, because the constructor's `createDocument` + * truncates whatever table is already there and sbt runs amber suites in parallel in one JVM + * (DefaultCostEstimatorSpec owns wid 1 / eid 1). They must also differ from each other, so a + * URI built from the wrong identity is a visibly different URI. + * + * Not covered on purpose: the catch arm around `runtimeStatsWriter.close()` (iceberg's close is + * a repeatable buffer flush and does not throw), and the catch arm in `storeRuntimeStatistics` + * (it runs on the private persist executor, which swallows anything a test could observe). + */ +class ExecutionStatsServiceSpec + extends TestKit(ActorSystem("ExecutionStatsServiceSpec")) + with AnyFlatSpecLike + with Matchers + with BeforeAndAfterAll + with MockTexeraDB { + + private val testUid: Integer = 9107 + private val testWid: Integer = 9107 + private val testEid: Integer = 9108 + + private val statsUri: URI = VFSURIFactory.createRuntimeStatisticsURI( + WorkflowIdentity(testWid.longValue()), + ExecutionIdentity(testEid.longValue()) + ) + + override protected def beforeAll(): Unit = { + initializeDBAndReplaceDSLContext() + + val user = new User + user.setUid(testUid) + user.setName("execution-stats-test-user") + user.setEmail(s"[email protected]") + new UserDao(getDSLContext.configuration()).insert(user) + + val workflow = new Workflow + workflow.setWid(testWid) + workflow.setName(s"execution-stats-test-$testWid") + workflow.setContent("{}") + workflow.setDescription("") + workflow.setCreationTime(new Timestamp(System.currentTimeMillis())) + workflow.setLastModifiedTime(new Timestamp(System.currentTimeMillis())) + new WorkflowDao(getDSLContext.configuration()).insert(workflow) + + val version = new WorkflowVersion + version.setWid(testWid) + version.setContent("{}") + version.setCreationTime(new Timestamp(System.currentTimeMillis())) + new WorkflowVersionDao(getDSLContext.configuration()).insert(version) + + // The execution row is what the constructor stamps the statistics URI onto; it has to be + // reachable from testWid through workflow_version, which is how the update is scoped. + val execution = new WorkflowExecutions + execution.setEid(testEid) + execution.setVid(version.getVid) + execution.setUid(testUid) + execution.setStatus(0.toByte) + execution.setStartingTime(new Timestamp(System.currentTimeMillis())) + execution.setBookmarked(false) + execution.setName("execution-stats-test-execution") + execution.setEnvironmentVersion("test-env") + new WorkflowExecutionsDao(getDSLContext.configuration()).insert(execution) + } + + override protected def afterAll(): Unit = { + try { + TestKit.shutdownActorSystem(system) + closeConnectionPool() + } finally super.afterAll() + } + + /** Empty-plan client that captures every callback the service registers, keyed by event type. */ + private final class TestAmberClient + extends AmberClient( + system, + new WorkflowContext(), + PhysicalPlan(Set.empty, Set.empty), + CoordinatorConfig(None, None, None, None), + _ => () + ) { + var shutdownCount = 0 + private val callbacks = mutable.Map.empty[Class[_], Any => Unit] + + override def shutdown(): Unit = { + shutdownCount += 1 + super.shutdown() + } + + override def registerCallback[T](callback: T => Unit)(implicit ct: ClassTag[T]): Disposable = { + callbacks(ct.runtimeClass) = callback.asInstanceOf[Any => Unit] + Disposable.empty() + } + + /** Delivers an engine event the way the client's observable would. */ + def fire[T <: AnyRef](event: T)(implicit ct: ClassTag[T]): Unit = + callbacks(ct.runtimeClass)(event) + + def dispose(): Unit = super.shutdown() + } + + private final class Fixture( + val client: TestAmberClient, + val stateStore: ExecutionStateStore, + val service: ExecutionStatsService + ) { + def close(): Unit = { + service.unsubscribeAll() + client.dispose() + } + } + + private def withService(body: Fixture => Unit): Unit = { + val client = new TestAmberClient + val stateStore = new ExecutionStateStore + val service = new ExecutionStatsService( + client, + stateStore, + new WorkflowContext( + workflowId = WorkflowIdentity(testWid.longValue()), + executionId = ExecutionIdentity(testEid.longValue()) + ) + ) + val fixture = new Fixture(client, stateStore, service) + try body(fixture) + finally fixture.close() + } + + /** + * One batch of websocket events is published per state update, so a test that cares about + * what a *second* update publishes has to subscribe before the first one. Subscribing later + * would also replay the diff between the default state and the current one as a batch. + */ + private def recordBatches( + store: StateStore[ExecutionStatsStore] + )(body: => Unit): Seq[Iterable[TexeraWebSocketEvent]] = { + val batches = ListBuffer.empty[Iterable[TexeraWebSocketEvent]] + val subscription = store.getWebsocketEventObservable + .subscribe((batch: Iterable[TexeraWebSocketEvent]) => batches += batch) + try body + finally subscription.dispose() + batches.toSeq + } + + private def metrics( + state: WorkflowAggregatedState, + inputs: Seq[(Int, Long, Long)] = Seq.empty, + outputs: Seq[(Int, Long, Long)] = Seq.empty, + numWorkers: Int = 1, + dataProcessingTime: Long = 0, + controlProcessingTime: Long = 0, + idleTime: Long = 0 + ): OperatorMetrics = { + def portMetrics(ports: Seq[(Int, Long, Long)]): Seq[PortTupleMetricsMapping] = + ports.map { + case (port, count, size) => + PortTupleMetricsMapping(PortIdentity(port), TupleMetrics(count, size)) + } + + OperatorMetrics( + state, + OperatorStatistics( + portMetrics(inputs), + portMetrics(outputs), + numWorkers, + dataProcessingTime, + controlProcessingTime, + idleTime + ) + ) + } + + private def statisticsEvents( + batch: Iterable[TexeraWebSocketEvent] + ): Map[String, OperatorAggregatedMetrics] = + batch.collect { case e: OperatorStatisticsUpdateEvent => e.operatorStatistics }.flatten.toMap + + /** + * The persist callback hands its work to a private single-threaded executor. A no-op task + * queued behind it can only run once that work is done, which is the barrier a test needs + * before it closes the writer and reads the table back. + */ + private def awaitPersist(service: ExecutionStatsService): Unit = { + val field = classOf[ExecutionStatsService].getDeclaredField("metricsPersistThread") + field.setAccessible(true) + field.get(service).asInstanceOf[ExecutorService].submit((() => ()): Runnable).get() + } + + /** Reads the committed statistics rows. Opening (never re-creating) the document is what makes + * this a read: `createDocument` truncates. + */ + private def persistedRows(): Seq[Tuple] = + DocumentFactory + .openDocument(statsUri) + ._1 + .asInstanceOf[VirtualDocument[Tuple]] + .get() + .toSeq + + "the operator statistics callback" should + "publish per-port and aggregate metrics for every operator" in { + // Input and output are two symmetrical groups of fields on the same event; the frontend + // shows them as separate columns, so they are given distinct counts, sizes and port ids + // here to make crossing them visible. + withService { f => + val batches = recordBatches(f.stateStore.statsStore) { + f.client.fire( + ExecutionStatsUpdate( + Map( + "opA" -> metrics( + RUNNING, + inputs = Seq((0, 3L, 30L), (1, 4L, 40L)), + outputs = Seq((2, 5L, 50L)), + numWorkers = 2, + dataProcessingTime = 11L, + controlProcessingTime = 12L, + idleTime = 13L + ), + // A second operator, because the handler maps over every reported operator and one + // operator cannot show that: `operatorInfo.collect` could be reduced to + // `operatorInfo.take(1).collect` and a single-operator fixture would not notice. + "opB" -> metrics( + COMPLETED, + inputs = Seq((0, 9L, 90L)), + outputs = Seq((1, 8L, 80L)), + numWorkers = 4, + dataProcessingTime = 21L, + controlProcessingTime = 22L, + idleTime = 23L + ) + ) + ) + ) + } + + val published = statisticsEvents(batches.flatten) + published.keySet shouldBe Set("opA", "opB") + val opA = published("opA") + opA.operatorState shouldBe "Running" + opA.aggregatedInputRowCount shouldBe 7L + opA.aggregatedInputSize shouldBe 70L + opA.inputPortMetrics shouldBe Map("0" -> 3L, "1" -> 4L) + opA.aggregatedOutputRowCount shouldBe 5L + opA.aggregatedOutputSize shouldBe 50L + opA.outputPortMetrics shouldBe Map("2" -> 5L) + opA.numWorkers shouldBe 2L + opA.aggregatedDataProcessingTime shouldBe 11L + opA.aggregatedControlProcessingTime shouldBe 12L + opA.aggregatedIdleTime shouldBe 13L + + val opB = published("opB") + opB.operatorState shouldBe "Completed" + opB.aggregatedInputRowCount shouldBe 9L + opB.aggregatedOutputRowCount shouldBe 8L + opB.numWorkers shouldBe 4L + } + } + + it should "publish nothing when the reported metrics repeat" in { + // The frontend redraws the whole statistics panel per event, so re-announcing identical + // numbers is pure churn on every heartbeat of a long-running execution. + withService { f => + val reported = Map("opA" -> metrics(RUNNING, inputs = Seq((0, 1L, 10L)))) + + val batches = recordBatches(f.stateStore.statsStore) { + f.client.fire(ExecutionStatsUpdate(reported)) + // Same metrics, plus a change the duration handler reacts to, so a second batch is + // published at all -- it just must not carry a statistics event. + f.stateStore.statsStore.updateState(_.withOperatorInfo(reported).withStartTimeStamp(1L)) + } + + batches should have size 2 + statisticsEvents(batches.head).keySet shouldBe Set("opA") + // Collected by event TYPE, not by payload: flattening the payload map would treat a + // present-but-empty OperatorStatisticsUpdateEvent as "nothing published", when it is in fact + // churn on the websocket. This matches the two sibling "publish nothing" tests below. + batches.last.collect { case e: OperatorStatisticsUpdateEvent => e } shouldBe empty + } + } + + "the worker assignment callback" should + "publish one event per operator carrying that operator's workers" in { + withService { f => + val batches = recordBatches(f.stateStore.statsStore) { + f.client.fire( + WorkerAssignmentUpdate( + Map("opA" -> Seq("worker-a-0", "worker-a-1"), "opB" -> Seq("worker-b-0")) + ) + ) + } + + val assignments = batches.flatten.collect { + case e: WorkerAssignmentUpdateEvent => e.operatorId -> e.workerIds + } + assignments.toMap shouldBe Map( + "opA" -> Seq("worker-a-0", "worker-a-1"), + "opB" -> Seq("worker-b-0") + ) + } + } + + it should "publish nothing when the assignment is unchanged" in { + withService { f => + val batches = recordBatches(f.stateStore.statsStore) { + f.client.fire(WorkerAssignmentUpdate(Map("opA" -> Seq("worker-a-0")))) + // A later statistics update changes the state without touching the assignment. + f.client.fire(ExecutionStatsUpdate(Map("opA" -> metrics(RUNNING)))) + } + + batches should have size 2 + batches.head.collect { case e: WorkerAssignmentUpdateEvent => e } should have size 1 + batches.last.collect { case e: WorkerAssignmentUpdateEvent => e } shouldBe empty + } + } + + "the execution duration handler" should "report the time elapsed so far while running" in { + withService { f => + val startedAgo = 1500L + val before = System.currentTimeMillis() + val batches = recordBatches(f.stateStore.statsStore) { + f.stateStore.statsStore.updateState( + _.withStartTimeStamp(System.currentTimeMillis() - startedAgo) + ) + } + val elapsedDuringUpdate = System.currentTimeMillis() - before + + val durations = batches.flatten.collect { case e: ExecutionDurationUpdateEvent => e } + durations should have size 1 + durations.head.isRunning shouldBe true + // The frontend keeps ticking the clock forward from this value, so it is measured + // against now rather than against the (still unset) end timestamp. + // + // Bounded on BOTH sides on purpose. A lower bound alone is satisfied by any arithmetic that + // grows the number, so `currentTime - startTimeStamp` could become `currentTime + + // startTimeStamp` -- roughly 111 years -- and still pass. + durations.head.duration should be >= startedAgo + durations.head.duration should be <= startedAgo + elapsedDuringUpdate + } + } + + it should "report the total duration once the execution has ended" in { + withService { f => + val batches = recordBatches(f.stateStore.statsStore) { + f.stateStore.statsStore.updateState(_.withStartTimeStamp(1000L).withEndTimeStamp(4200L)) + } + + val durations = batches.flatten.collect { case e: ExecutionDurationUpdateEvent => e } + durations should have size 1 + durations.head.duration shouldBe 3200L + durations.head.isRunning shouldBe false + } + } + + it should "publish nothing when neither timestamp moved" in { + withService { f => + val batches = recordBatches(f.stateStore.statsStore) { + f.stateStore.statsStore.updateState(_.withStartTimeStamp(1000L)) + f.client.fire(ExecutionStatsUpdate(Map("opA" -> metrics(RUNNING)))) + } + + batches should have size 2 + batches.head.collect { case e: ExecutionDurationUpdateEvent => e } should have size 1 + batches.last.collect { case e: ExecutionDurationUpdateEvent => e } shouldBe empty + } + } + + "the recovery status callback" should "mirror the reported flag onto the metadata store" in { + withService { f => + f.client.fire(WorkflowRecoveryStatus(isRecovering = true)) + f.stateStore.metadataStore.getState.isRecovering shouldBe true + + // Both directions: the frontend leaves its recovery banner up until this flips back. + f.client.fire(WorkflowRecoveryStatus(isRecovering = false)) + f.stateStore.metadataStore.getState.isRecovering shouldBe false + } + } + + "a fatal error" should "stop the client and record the failure against the reporting operator" in { + withService { f => + val cause = new RuntimeException("stats-service-boom") + + f.client.fire(FatalError(cause, Some(ActorVirtualIdentity("Worker:WF1-udf1-main-0")))) + + // The engine is already lost when it reports a fatal error; leaving the client running + // would keep the execution's actors alive behind a failed workflow. + f.client.shutdownCount shouldBe 1 + f.stateStore.statsStore.getState.endTimeStamp should be > 0L + f.stateStore.metadataStore.getState.state shouldBe FAILED + + val errors = f.stateStore.metadataStore.getState.fatalErrors + errors should have size 1 + errors.head.`type` shouldBe EXECUTION_FAILURE + // The error panel groups by operator and links to the worker; the two are separate + // fields derived from the same actor id. + errors.head.operatorId shouldBe "udf1" + errors.head.workerId shouldBe "Worker:WF1-udf1-main-0" + errors.head.message shouldBe cause.toString + errors.head.details should include("stats-service-boom") + } + } + + "the service constructor" should "stamp the runtime statistics URI onto the execution row" in { + // Spelled out rather than rebuilt from VFSURIFactory: the URI is what the dashboard later + // opens, so a wid/eid mix-up has to show up as a different string here. + getDSLContext + .update(WORKFLOW_EXECUTIONS) + .setNull(WORKFLOW_EXECUTIONS.RUNTIME_STATS_URI) + .where(WORKFLOW_EXECUTIONS.EID.eq(testEid)) + .execute() + + withService { _ => + getDSLContext + .select(WORKFLOW_EXECUTIONS.RUNTIME_STATS_URI) + .from(WORKFLOW_EXECUTIONS) + .where(WORKFLOW_EXECUTIONS.EID.eq(testEid)) + .fetchOne(WORKFLOW_EXECUTIONS.RUNTIME_STATS_URI) shouldBe + s"vfs:///wid/$testWid/eid/$testEid/runtimestatistics" + } + } + + "the persistence callback" should + "carry an operator that stopped reporting into the persisted statistics" in { + withService { f => + val opBFinal = metrics( + COMPLETED, + inputs = Seq((0, 3L, 30L)), + outputs = Seq((1, 4L, 40L)), + numWorkers = 2, + dataProcessingTime = 7L, + controlProcessingTime = 8L, + idleTime = 9L + ) + + f.client.fire( + RuntimeStatisticsPersist( + Map("opA" -> metrics(RUNNING, inputs = Seq((0, 1L, 10L))), "opB" -> opBFinal) + ) + ) + awaitPersist(f.service) + // opB is gone from the second report, as happens once an operator completes and stops + // being polled. + f.client.fire(RuntimeStatisticsPersist(Map("opA" -> metrics(COMPLETED)))) + awaitPersist(f.service) + f.client.fire(ExecutionStateUpdate(COMPLETED)) + + val rows = persistedRows() + rows.map(_.getField[String]("operatorId")) should + contain theSameElementsAs Seq("opA", "opA", "opB", "opB") + + // The dashboard replays these rows as a per-operator time series, so an operator that + // goes quiet has to keep contributing its last known numbers instead of dropping out of + // the chart at that point. + val opBRows = rows.filter(_.getField[String]("operatorId") == "opB") + opBRows should have size 2 + opBRows.foreach { row => + row.getField[Long]("inputTupleCnt") shouldBe 3L + row.getField[Long]("inputTupleSize") shouldBe 30L + row.getField[Long]("outputTupleCnt") shouldBe 4L + row.getField[Long]("outputTupleSize") shouldBe 40L + row.getField[Long]("dataProcessingTime") shouldBe 7L + row.getField[Long]("controlProcessingTime") shouldBe 8L + row.getField[Long]("idleTime") shouldBe 9L + row.getField[Int]("numWorkers") shouldBe 2 + // COMPLETED, as a status code -- the column the dashboard colours the row by. + row.getField[Int]("status") shouldBe 3 + } + } + } + + it should "hold the statistics back until the execution reaches a terminal state" in { + // All three terminal states are meant to trigger the commit, and an execution reaches only + // one of them, so each gets its own service -- and with it a fresh statistics table. + Seq(COMPLETED, FAILED, KILLED).foreach { terminalState => + withClue(s"terminal state $terminalState: ") { + withService { f => + f.client.fire(RuntimeStatisticsPersist(Map("opA" -> metrics(RUNNING)))) + awaitPersist(f.service) + + // Mid-execution state changes must not commit: the writer is closed once, and closing + // it early would strand every later statistics row in the buffer. + f.client.fire(ExecutionStateUpdate(RUNNING)) + persistedRows() shouldBe empty + + f.client.fire(ExecutionStateUpdate(terminalState)) + persistedRows() should have size 1 + } + } + } + } +}
