You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@texera.apache.org by "aglinxinyuan (via GitHub)" <gi...@apache.org> on 2026/07/05 09:12:50 UTC
[PR] test(amber): add EmbeddedControlMessageManager alignment tests [texera]
aglinxinyuan opened a new pull request, #6158:
URL: https://github.com/apache/texera/pull/6158
### What changes were proposed in this PR?
Adds `EmbeddedControlMessageManagerSpec` (new) for the amber worker's `EmbeddedControlMessageManager`, which had no spec. Using directly-constructed collaborators (`NetworkInputGateway` + `InputManager`, no actor system), it covers the `isECMAligned` / `getChannelsWithinScope` branches:
- `ALL_ALIGNMENT` (empty scope) — not aligned until every data channel has delivered the ECM;
- `PORT_ALIGNMENT` — aligns on the port's channel set;
- `NO_ALIGNMENT` — triggers only on the first ECM;
- an unsupported ECM type throws `IllegalArgumentException`;
- the non-empty-scope filter path.
No production code changed.
### Any related issues, documentation, discussions?
Coverage gaps identified from the Codecov report for `apache/texera` (amber module coverage push toward 80%).
### How was this PR tested?
```
sbt "WorkflowExecutionService/testOnly *EmbeddedControlMessageManagerSpec"
```
All 5 pass. `scalafmt` and `scalafixAll --check` are clean.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add EmbeddedControlMessageManager alignment tests [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #6158:
URL: https://github.com/apache/texera/pull/6158#discussion_r3524610869
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/EmbeddedControlMessageManagerSpec.scala:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.architecture.worker
+
+import org.apache.texera.amber.core.tuple.Schema
+import org.apache.texera.amber.core.virtualidentity.{
+ ActorVirtualIdentity,
+ ChannelIdentity,
+ EmbeddedControlMessageIdentity,
+ OperatorIdentity,
+ PhysicalOpIdentity
+}
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.core.workflow.WorkflowContext.DEFAULT_WORKFLOW_ID
+import org.apache.texera.amber.engine.architecture.messaginglayer.{
+ InputManager,
+ NetworkInputGateway
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{
+ EmbeddedControlMessage,
+ EmbeddedControlMessageType
+}
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.DPInputQueueElement
+import org.apache.texera.amber.util.VirtualIdentityUtils
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.util.concurrent.LinkedBlockingQueue
+
+class EmbeddedControlMessageManagerSpec extends AnyFlatSpec with Matchers {
+
+ private val testOpId = PhysicalOpIdentity(OperatorIdentity("testop"), "main")
+ private val upstreamOpIdA = PhysicalOpIdentity(OperatorIdentity("senderA"), "main")
+ private val upstreamOpIdB = PhysicalOpIdentity(OperatorIdentity("senderB"), "main")
+
+ private val actorId: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, testOpId, 0)
+ private val senderA: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdA, 0)
+ private val senderB: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdB, 0)
+
+ private def mkGateway: NetworkInputGateway = new NetworkInputGateway(actorId)
+
+ private def mkInputManager: InputManager =
+ new InputManager(actorId, new LinkedBlockingQueue[DPInputQueueElement]())
+
+ private def mkEcm(
+ ecmType: EmbeddedControlMessageType,
+ scope: Seq[ChannelIdentity]
+ ): EmbeddedControlMessage =
+ EmbeddedControlMessage(
+ EmbeddedControlMessageIdentity("ecm"),
+ ecmType,
+ scope,
+ Map.empty
+ )
+
+ "EmbeddedControlMessageManager" should "align ALL_ALIGNMENT only after every data channel is received" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chA = ChannelIdentity(senderA, actorId, isControl = false)
+ val chB = ChannelIdentity(senderB, actorId, isControl = false)
+ gateway.getChannel(chA).setPortId(portId)
+ gateway.getChannel(chB).setPortId(portId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.ALL_ALIGNMENT, Seq.empty)
+
+ // first data channel: not yet received from all channels in scope
+ mgr.isECMAligned(chA, ecm) shouldBe false
+ // second (last) data channel: now aligned
+ mgr.isECMAligned(chB, ecm) shouldBe true
+ }
+
+ it should "align PORT_ALIGNMENT once all channels registered on the port are received" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chId = ChannelIdentity(senderA, actorId, isControl = false)
+ gateway.getChannel(chId).setPortId(portId)
+
+ inputManager.addPort(portId, Schema(), List.empty, List.empty)
+ inputManager.getPort(portId).channels.add(chId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.PORT_ALIGNMENT, Seq(chId))
+
+ mgr.isECMAligned(chId, ecm) shouldBe true
+ }
Review Comment:
The PORT_ALIGNMENT test currently only has a single channel registered on the port, so it can’t exercise the “wait until *all* port channels have delivered the ECM” behavior (it will always align on the first receipt). Adding a second channel on the same port and asserting false-then-true would better cover the intended branch semantics.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/EmbeddedControlMessageManagerSpec.scala:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.architecture.worker
+
+import org.apache.texera.amber.core.tuple.Schema
+import org.apache.texera.amber.core.virtualidentity.{
+ ActorVirtualIdentity,
+ ChannelIdentity,
+ EmbeddedControlMessageIdentity,
+ OperatorIdentity,
+ PhysicalOpIdentity
+}
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.core.workflow.WorkflowContext.DEFAULT_WORKFLOW_ID
+import org.apache.texera.amber.engine.architecture.messaginglayer.{
+ InputManager,
+ NetworkInputGateway
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{
+ EmbeddedControlMessage,
+ EmbeddedControlMessageType
+}
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.DPInputQueueElement
+import org.apache.texera.amber.util.VirtualIdentityUtils
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.util.concurrent.LinkedBlockingQueue
+
+class EmbeddedControlMessageManagerSpec extends AnyFlatSpec with Matchers {
+
+ private val testOpId = PhysicalOpIdentity(OperatorIdentity("testop"), "main")
+ private val upstreamOpIdA = PhysicalOpIdentity(OperatorIdentity("senderA"), "main")
+ private val upstreamOpIdB = PhysicalOpIdentity(OperatorIdentity("senderB"), "main")
+
+ private val actorId: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, testOpId, 0)
+ private val senderA: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdA, 0)
+ private val senderB: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdB, 0)
+
+ private def mkGateway: NetworkInputGateway = new NetworkInputGateway(actorId)
+
+ private def mkInputManager: InputManager =
+ new InputManager(actorId, new LinkedBlockingQueue[DPInputQueueElement]())
+
+ private def mkEcm(
+ ecmType: EmbeddedControlMessageType,
+ scope: Seq[ChannelIdentity]
+ ): EmbeddedControlMessage =
+ EmbeddedControlMessage(
+ EmbeddedControlMessageIdentity("ecm"),
+ ecmType,
+ scope,
+ Map.empty
+ )
+
+ "EmbeddedControlMessageManager" should "align ALL_ALIGNMENT only after every data channel is received" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chA = ChannelIdentity(senderA, actorId, isControl = false)
+ val chB = ChannelIdentity(senderB, actorId, isControl = false)
+ gateway.getChannel(chA).setPortId(portId)
+ gateway.getChannel(chB).setPortId(portId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.ALL_ALIGNMENT, Seq.empty)
+
+ // first data channel: not yet received from all channels in scope
+ mgr.isECMAligned(chA, ecm) shouldBe false
+ // second (last) data channel: now aligned
+ mgr.isECMAligned(chB, ecm) shouldBe true
+ }
+
+ it should "align PORT_ALIGNMENT once all channels registered on the port are received" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chId = ChannelIdentity(senderA, actorId, isControl = false)
+ gateway.getChannel(chId).setPortId(portId)
+
+ inputManager.addPort(portId, Schema(), List.empty, List.empty)
+ inputManager.getPort(portId).channels.add(chId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.PORT_ALIGNMENT, Seq(chId))
+
+ mgr.isECMAligned(chId, ecm) shouldBe true
+ }
+
+ it should "align NO_ALIGNMENT only on the first received ECM" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chA = ChannelIdentity(senderA, actorId, isControl = false)
+ val chB = ChannelIdentity(senderB, actorId, isControl = false)
+ gateway.getChannel(chA).setPortId(portId)
+ gateway.getChannel(chB).setPortId(portId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.NO_ALIGNMENT, Seq.empty)
+
+ // first ECM triggers
+ mgr.isECMAligned(chA, ecm) shouldBe true
+ // subsequent ECM does not trigger again
+ mgr.isECMAligned(chB, ecm) shouldBe false
+ }
+
+ it should "throw IllegalArgumentException for an unsupported ECM type" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chId = ChannelIdentity(senderA, actorId, isControl = false)
+ gateway.getChannel(chId).setPortId(portId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.Unrecognized(-1), Seq.empty)
+
+ assertThrows[IllegalArgumentException] {
+ mgr.isECMAligned(chId, ecm)
+ }
+ }
+
+ it should "restrict the scope to channels targeting this worker when scope is non-empty" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ // an in-scope channel targeting this worker, plus an out-of-scope extra channel
+ val inScope = ChannelIdentity(senderA, actorId, isControl = false)
+ val extra = ChannelIdentity(senderB, actorId, isControl = false)
+ gateway.getChannel(inScope).setPortId(portId)
+ gateway.getChannel(extra).setPortId(portId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ // scope names only the in-scope channel (toWorkerId == actorId), so extra is ignored
+ val ecm = mkEcm(EmbeddedControlMessageType.ALL_ALIGNMENT, Seq(inScope))
+
+ mgr.isECMAligned(inScope, ecm) shouldBe true
+ }
Review Comment:
The non-empty-scope test enters the `ecm.scope.nonEmpty` branch, but it doesn’t validate the key behavior of filtering out scope entries that *don’t* target this worker (`toWorkerId != actorId`). Including such a channel in the scope (and registering it in the gateway) would make the test actually assert the filter semantics.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add EmbeddedControlMessageManager alignment tests [texera]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #6158:
URL: https://github.com/apache/texera/pull/6158#issuecomment-4885528102
## [Codecov](https://app.codecov.io/gh/apache/texera/pull/6158?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 62.23%. Comparing base ([`5acb734`](https://app.codecov.io/gh/apache/texera/commit/5acb73433a17ebfaba73d5da19b00b4fc2d3823f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`b154252`](https://app.codecov.io/gh/apache/texera/commit/b154252aefeb8f8e3ed09df24c42a821fa8c13e7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on main.
:white_check_mark: All tests successful. No failed tests found.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #6158 +/- ##
============================================
+ Coverage 59.85% 62.23% +2.38%
+ Complexity 3302 2176 -1126
============================================
Files 1131 861 -270
Lines 43770 35428 -8342
Branches 4745 3602 -1143
============================================
- Hits 26197 22048 -4149
+ Misses 16147 12384 -3763
+ Partials 1426 996 -430
```
| [Flag](https://app.codecov.io/gh/apache/texera/pull/6158/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | | *Carryforward flag |
|---|---|---|---|
| [access-control-service](https://app.codecov.io/gh/apache/texera/pull/6158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.00% <ø> (ø)` | | Carriedforward from [5acb734](https://app.codecov.io/gh/apache/texera/commit/5acb73433a17ebfaba73d5da19b00b4fc2d3823f?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [agent-service](https://app.codecov.io/gh/apache/texera/pull/6158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `44.59% <ø> (ø)` | | Carriedforward from [5acb734](https://app.codecov.io/gh/apache/texera/commit/5acb73433a17ebfaba73d5da19b00b4fc2d3823f?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [amber](https://app.codecov.io/gh/apache/texera/pull/6158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `79.89% <ø> (+14.98%)` | :arrow_up: | |
| [computing-unit-managing-service](https://app.codecov.io/gh/apache/texera/pull/6158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | | Carriedforward from [5acb734](https://app.codecov.io/gh/apache/texera/commit/5acb73433a17ebfaba73d5da19b00b4fc2d3823f?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [config-service](https://app.codecov.io/gh/apache/texera/pull/6158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `52.30% <ø> (ø)` | | Carriedforward from [5acb734](https://app.codecov.io/gh/apache/texera/commit/5acb73433a17ebfaba73d5da19b00b4fc2d3823f?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [file-service](https://app.codecov.io/gh/apache/texera/pull/6158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.81% <ø> (ø)` | | Carriedforward from [5acb734](https://app.codecov.io/gh/apache/texera/commit/5acb73433a17ebfaba73d5da19b00b4fc2d3823f?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [frontend](https://app.codecov.io/gh/apache/texera/pull/6158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `51.48% <ø> (ø)` | | Carriedforward from [5acb734](https://app.codecov.io/gh/apache/texera/commit/5acb73433a17ebfaba73d5da19b00b4fc2d3823f?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [notebook-migration-service](https://app.codecov.io/gh/apache/texera/pull/6158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.57% <ø> (ø)` | | Carriedforward from [5acb734](https://app.codecov.io/gh/apache/texera/commit/5acb73433a17ebfaba73d5da19b00b4fc2d3823f?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [pyamber](https://app.codecov.io/gh/apache/texera/pull/6158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.18% <ø> (ø)` | | Carriedforward from [5acb734](https://app.codecov.io/gh/apache/texera/commit/5acb73433a17ebfaba73d5da19b00b4fc2d3823f?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [workflow-compiling-service](https://app.codecov.io/gh/apache/texera/pull/6158/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.14% <ø> (ø)` | | Carriedforward from [5acb734](https://app.codecov.io/gh/apache/texera/commit/5acb73433a17ebfaba73d5da19b00b4fc2d3823f?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
*This pull request uses carry forward flags. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/texera/pull/6158?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add EmbeddedControlMessageManager alignment tests [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6158:
URL: https://github.com/apache/texera/pull/6158#issuecomment-4885514887
<!-- texera-benchmarks-comment -->
## ⚠️ Benchmark changes need a look
🟢 2 better · 🔴 1 worse · ⚪ 12 noise (<±5%) · 0 without baseline
> Compared against main `5acb734` benchmarked on **this same runner**, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.
[Dashboard](https://apache.github.io/texera/dev/bench/) · [Run](https://github.com/apache/texera/actions/runs/28735872322)
| | config | throughput | MB/s | latency | max Δ latest / 7d |
|---|---|---:|---:|---:|---:|
| 🔴 | bs=10 sw=10 sl=64 | 433 | 0.264 | 22,845/33,076/33,076 us | 🔴 +16.3% / 🔴 +118.4% |
| 🟢 | bs=100 sw=10 sl=64 | 932 | 0.569 | 103,121/153,204/153,204 us | 🟢 -11.4% / 🔴 +41.9% |
| ⚪ | bs=1000 sw=10 sl=64 | 1,117 | 0.682 | 895,109/919,757/919,757 us | ⚪ within ±5% / 🟢 -13.2% |
<details><summary>Baseline details</summary>
Latest main `5acb734` from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---:|---:|---:|---:|---:|
| bs=10 sw=10 sl=64 | throughput | 433 tuples/sec | 453 tuples/sec | 776.36 tuples/sec | -4.4% | -44.2% |
| bs=10 sw=10 sl=64 | MB/s | 0.264 MB/s | 0.276 MB/s | 0.474 MB/s | -4.3% | -44.3% |
| bs=10 sw=10 sl=64 | p50 | 22,845 us | 19,647 us | 12,636 us | +16.3% | +80.8% |
| bs=10 sw=10 sl=64 | p95 | 33,076 us | 32,811 us | 15,143 us | +0.8% | +118.4% |
| bs=10 sw=10 sl=64 | p99 | 33,076 us | 32,811 us | 18,954 us | +0.8% | +74.5% |
| bs=100 sw=10 sl=64 | throughput | 932 tuples/sec | 943 tuples/sec | 985.33 tuples/sec | -1.2% | -5.4% |
| bs=100 sw=10 sl=64 | MB/s | 0.569 MB/s | 0.576 MB/s | 0.601 MB/s | -1.2% | -5.4% |
| bs=100 sw=10 sl=64 | p50 | 103,121 us | 101,023 us | 101,671 us | +2.1% | +1.4% |
| bs=100 sw=10 sl=64 | p95 | 153,204 us | 172,961 us | 107,939 us | -11.4% | +41.9% |
| bs=100 sw=10 sl=64 | p99 | 153,204 us | 172,961 us | 113,798 us | -11.4% | +34.6% |
| bs=1000 sw=10 sl=64 | throughput | 1,117 tuples/sec | 1,103 tuples/sec | 1,016 tuples/sec | +1.3% | +9.9% |
| bs=1000 sw=10 sl=64 | MB/s | 0.682 MB/s | 0.673 MB/s | 0.62 MB/s | +1.3% | +9.9% |
| bs=1000 sw=10 sl=64 | p50 | 895,109 us | 904,735 us | 989,693 us | -1.1% | -9.6% |
| bs=1000 sw=10 sl=64 | p95 | 919,757 us | 936,769 us | 1,028,327 us | -1.8% | -10.6% |
| bs=1000 sw=10 sl=64 | p99 | 919,757 us | 936,769 us | 1,059,969 us | -1.8% | -13.2% |
</details>
<details><summary>Raw CSV</summary>
```csv
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,461.73,200,128000,433,0.264,22844.76,33076.27,33076.27
1,100,10,64,20,2146.68,2000,1280000,932,0.569,103121.24,153203.63,153203.63
2,1000,10,64,20,17903.93,20000,12800000,1117,0.682,895109.07,919757.20,919757.20
```
</details>
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add EmbeddedControlMessageManager alignment tests [texera]
Posted by "xuang7 (via GitHub)" <gi...@apache.org>.
xuang7 merged PR #6158:
URL: https://github.com/apache/texera/pull/6158
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add EmbeddedControlMessageManager alignment tests [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #6158:
URL: https://github.com/apache/texera/pull/6158#discussion_r3531519569
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/EmbeddedControlMessageManagerSpec.scala:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.architecture.worker
+
+import org.apache.texera.amber.core.tuple.Schema
+import org.apache.texera.amber.core.virtualidentity.{
+ ActorVirtualIdentity,
+ ChannelIdentity,
+ EmbeddedControlMessageIdentity,
+ OperatorIdentity,
+ PhysicalOpIdentity
+}
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.core.workflow.WorkflowContext.DEFAULT_WORKFLOW_ID
+import org.apache.texera.amber.engine.architecture.messaginglayer.{
+ InputManager,
+ NetworkInputGateway
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{
+ EmbeddedControlMessage,
+ EmbeddedControlMessageType
+}
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.DPInputQueueElement
+import org.apache.texera.amber.util.VirtualIdentityUtils
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.util.concurrent.LinkedBlockingQueue
+
+class EmbeddedControlMessageManagerSpec extends AnyFlatSpec with Matchers {
+
+ private val testOpId = PhysicalOpIdentity(OperatorIdentity("testop"), "main")
+ private val upstreamOpIdA = PhysicalOpIdentity(OperatorIdentity("senderA"), "main")
+ private val upstreamOpIdB = PhysicalOpIdentity(OperatorIdentity("senderB"), "main")
+
+ private val actorId: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, testOpId, 0)
+ private val senderA: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdA, 0)
+ private val senderB: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdB, 0)
+
+ private def mkGateway: NetworkInputGateway = new NetworkInputGateway(actorId)
+
+ private def mkInputManager: InputManager =
+ new InputManager(actorId, new LinkedBlockingQueue[DPInputQueueElement]())
+
+ private def mkEcm(
+ ecmType: EmbeddedControlMessageType,
+ scope: Seq[ChannelIdentity]
+ ): EmbeddedControlMessage =
+ EmbeddedControlMessage(
+ EmbeddedControlMessageIdentity("ecm"),
+ ecmType,
+ scope,
+ Map.empty
+ )
+
+ "EmbeddedControlMessageManager" should "align ALL_ALIGNMENT only after every data channel is received" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chA = ChannelIdentity(senderA, actorId, isControl = false)
+ val chB = ChannelIdentity(senderB, actorId, isControl = false)
+ gateway.getChannel(chA).setPortId(portId)
+ gateway.getChannel(chB).setPortId(portId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.ALL_ALIGNMENT, Seq.empty)
+
+ // first data channel: not yet received from all channels in scope
+ mgr.isECMAligned(chA, ecm) shouldBe false
+ // second (last) data channel: now aligned
+ mgr.isECMAligned(chB, ecm) shouldBe true
+ }
+
+ it should "align PORT_ALIGNMENT once all channels registered on the port are received" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chId = ChannelIdentity(senderA, actorId, isControl = false)
+ gateway.getChannel(chId).setPortId(portId)
+
+ inputManager.addPort(portId, Schema(), List.empty, List.empty)
+ inputManager.getPort(portId).channels.add(chId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.PORT_ALIGNMENT, Seq(chId))
+
+ mgr.isECMAligned(chId, ecm) shouldBe true
+ }
Review Comment:
Good catch — updated to register two channels on the port and assert false-then-true (`chA` then `chB`), so the "wait until all port channels have delivered" branch is now actually exercised instead of aligning on first receipt. Fixed in ec86000927.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/EmbeddedControlMessageManagerSpec.scala:
##########
@@ -0,0 +1,164 @@
+/*
+ * 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.architecture.worker
+
+import org.apache.texera.amber.core.tuple.Schema
+import org.apache.texera.amber.core.virtualidentity.{
+ ActorVirtualIdentity,
+ ChannelIdentity,
+ EmbeddedControlMessageIdentity,
+ OperatorIdentity,
+ PhysicalOpIdentity
+}
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.core.workflow.WorkflowContext.DEFAULT_WORKFLOW_ID
+import org.apache.texera.amber.engine.architecture.messaginglayer.{
+ InputManager,
+ NetworkInputGateway
+}
+import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{
+ EmbeddedControlMessage,
+ EmbeddedControlMessageType
+}
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.DPInputQueueElement
+import org.apache.texera.amber.util.VirtualIdentityUtils
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.util.concurrent.LinkedBlockingQueue
+
+class EmbeddedControlMessageManagerSpec extends AnyFlatSpec with Matchers {
+
+ private val testOpId = PhysicalOpIdentity(OperatorIdentity("testop"), "main")
+ private val upstreamOpIdA = PhysicalOpIdentity(OperatorIdentity("senderA"), "main")
+ private val upstreamOpIdB = PhysicalOpIdentity(OperatorIdentity("senderB"), "main")
+
+ private val actorId: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, testOpId, 0)
+ private val senderA: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdA, 0)
+ private val senderB: ActorVirtualIdentity =
+ VirtualIdentityUtils.createWorkerIdentity(DEFAULT_WORKFLOW_ID, upstreamOpIdB, 0)
+
+ private def mkGateway: NetworkInputGateway = new NetworkInputGateway(actorId)
+
+ private def mkInputManager: InputManager =
+ new InputManager(actorId, new LinkedBlockingQueue[DPInputQueueElement]())
+
+ private def mkEcm(
+ ecmType: EmbeddedControlMessageType,
+ scope: Seq[ChannelIdentity]
+ ): EmbeddedControlMessage =
+ EmbeddedControlMessage(
+ EmbeddedControlMessageIdentity("ecm"),
+ ecmType,
+ scope,
+ Map.empty
+ )
+
+ "EmbeddedControlMessageManager" should "align ALL_ALIGNMENT only after every data channel is received" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chA = ChannelIdentity(senderA, actorId, isControl = false)
+ val chB = ChannelIdentity(senderB, actorId, isControl = false)
+ gateway.getChannel(chA).setPortId(portId)
+ gateway.getChannel(chB).setPortId(portId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.ALL_ALIGNMENT, Seq.empty)
+
+ // first data channel: not yet received from all channels in scope
+ mgr.isECMAligned(chA, ecm) shouldBe false
+ // second (last) data channel: now aligned
+ mgr.isECMAligned(chB, ecm) shouldBe true
+ }
+
+ it should "align PORT_ALIGNMENT once all channels registered on the port are received" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chId = ChannelIdentity(senderA, actorId, isControl = false)
+ gateway.getChannel(chId).setPortId(portId)
+
+ inputManager.addPort(portId, Schema(), List.empty, List.empty)
+ inputManager.getPort(portId).channels.add(chId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.PORT_ALIGNMENT, Seq(chId))
+
+ mgr.isECMAligned(chId, ecm) shouldBe true
+ }
+
+ it should "align NO_ALIGNMENT only on the first received ECM" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chA = ChannelIdentity(senderA, actorId, isControl = false)
+ val chB = ChannelIdentity(senderB, actorId, isControl = false)
+ gateway.getChannel(chA).setPortId(portId)
+ gateway.getChannel(chB).setPortId(portId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.NO_ALIGNMENT, Seq.empty)
+
+ // first ECM triggers
+ mgr.isECMAligned(chA, ecm) shouldBe true
+ // subsequent ECM does not trigger again
+ mgr.isECMAligned(chB, ecm) shouldBe false
+ }
+
+ it should "throw IllegalArgumentException for an unsupported ECM type" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ val chId = ChannelIdentity(senderA, actorId, isControl = false)
+ gateway.getChannel(chId).setPortId(portId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ val ecm = mkEcm(EmbeddedControlMessageType.Unrecognized(-1), Seq.empty)
+
+ assertThrows[IllegalArgumentException] {
+ mgr.isECMAligned(chId, ecm)
+ }
+ }
+
+ it should "restrict the scope to channels targeting this worker when scope is non-empty" in {
+ val gateway = mkGateway
+ val inputManager = mkInputManager
+ val portId = PortIdentity()
+
+ // an in-scope channel targeting this worker, plus an out-of-scope extra channel
+ val inScope = ChannelIdentity(senderA, actorId, isControl = false)
+ val extra = ChannelIdentity(senderB, actorId, isControl = false)
+ gateway.getChannel(inScope).setPortId(portId)
+ gateway.getChannel(extra).setPortId(portId)
+
+ val mgr = new EmbeddedControlMessageManager(actorId, gateway, inputManager)
+ // scope names only the in-scope channel (toWorkerId == actorId), so extra is ignored
+ val ecm = mkEcm(EmbeddedControlMessageType.ALL_ALIGNMENT, Seq(inScope))
+
+ mgr.isECMAligned(inScope, ecm) shouldBe true
+ }
Review Comment:
Done. The scope now includes a `foreign` channel whose `toWorkerId != actorId` (registered in the gateway), so the `_.toWorkerId == actorId` filter in `getChannelsWithinScope` is genuinely asserted: alignment is reached on `inScope` alone precisely because `foreign` is filtered out. Fixed in ec86000927.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(amber): add EmbeddedControlMessageManager alignment tests [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6158:
URL: https://github.com/apache/texera/pull/6158#issuecomment-4885502800
<!-- texera-reviewer-suggestion -->
### Automated Reviewer Suggestions
Based on the `git blame` history of the changed files, we recommend the following reviewers:
- No candidates found from `git blame` history.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org