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/06/07 04:42:47 UTC
[PR] test(amber): add unit test coverage for InputManager [texera]
aglinxinyuan opened a new pull request, #5451:
URL: https://github.com/apache/texera/pull/5451
Closes #5450.
## Summary
Pin behavior of `InputManager` — the per-worker input broker that was the only previously-uncovered module in the `messaginglayer` package (every sibling — `WorkerPort`, `OutputManager`, `AmberFIFOChannel`, `CongestionControl`, `FlowControl`, `NetworkInputGateway`, `OrderingEnforcer`, `RangeBasedShuffle` — already has a spec).
| Spec | Source class | Tests |
| --- | --- | --- |
| `InputManagerSpec` | `InputManager` | 15 |
Spec file name follows the `<srcClassName>Spec.scala` one-to-one convention.
## Behavior pinned
| Surface | Contract |
| --- | --- |
| `getAllPorts` (fresh) | empty `Set` |
| `addPort(portId, schema, Nil, Nil)` | registers a fresh `WorkerPort` with the supplied schema; no reader threads attached |
| second `addPort` for the same `portId` | silently no-ops (first call wins; the prior `WorkerPort` survives, including any mutation to `completed`) |
| `addPort` with mismatched URI / partitioning sizes | throws `AssertionError` |
| `getPort(portId)` | is a pure lookup — successive calls return the same `WorkerPort` instance |
| `getInputPortReaderThreads` | empty `Map` when every port was added with `urisToRead = List.empty` |
| `hasUnfinishedInput` (fresh) | `false` (no batch initialized) |
| `getCurrentTuple` (fresh) | `null` |
| `getCurrentTuple` (empty batch) | `null` |
| `hasUnfinishedInput` (empty batch) | `false` |
| `initBatch(channelId, batch)` | replaces `inputBatch`, resets the cursor, surfaces the channel id via `currentChannelId` |
| `getNextTuple` over a batch | yields each element in order; `getCurrentTuple` tracks the last yielded value; `hasUnfinishedInput` flips to `false` once the last element is consumed |
| `initBatch` called a second time | replaces the prior batch entirely and restarts the cursor (proven by yielding the new batch's first element fresh) |
| `isPortCompleted` (non-materialized) | reads through to `WorkerPort.completed` — `false` initially, `true` once flipped |
| `startInputPortReaderThreads` (no threads) | safe no-op |
## Notes
Materialization-reader-thread paths (non-empty `urisToRead`) are deliberately left out of this characterization. Constructing an `InputPortMaterializationReaderThread` eagerly evaluates `toPartitioner(partitioning, workerActorId)` and pulls in document-factory plumbing that is out of scope for a unit spec — those code paths are exercised in higher-level worker / replay tests.
## Test plan
- [x] `WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.messaginglayer.InputManagerSpec` — 15 tests, all green
- [x] `scalafmtCheckAll` — clean
- [ ] CI to confirm
--
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 unit test coverage for InputManager [texera]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #5451:
URL: https://github.com/apache/texera/pull/5451#issuecomment-4641463384
## [Codecov](https://app.codecov.io/gh/apache/texera/pull/5451?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 51.90%. Comparing base ([`f2f0564`](https://app.codecov.io/gh/apache/texera/commit/f2f056455d850f6d4f99d81ac7e128fb74906a44?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`d863ec8`](https://app.codecov.io/gh/apache/texera/commit/d863ec8d468b35265d87a7cd55aa974b7b0859c2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 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 #5451 +/- ##
============================================
+ Coverage 51.85% 51.90% +0.04%
- Complexity 2468 2472 +4
============================================
Files 1067 1067
Lines 41258 41258
Branches 4437 4437
============================================
+ Hits 21394 21414 +20
+ Misses 18607 18580 -27
- Partials 1257 1264 +7
```
| [Flag](https://app.codecov.io/gh/apache/texera/pull/5451/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/5451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `42.22% <ø> (ø)` | | Carriedforward from [f2f0564](https://app.codecov.io/gh/apache/texera/commit/f2f056455d850f6d4f99d81ac7e128fb74906a44?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/5451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.76% <ø> (ø)` | | Carriedforward from [f2f0564](https://app.codecov.io/gh/apache/texera/commit/f2f056455d850f6d4f99d81ac7e128fb74906a44?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/5451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `52.88% <ø> (+0.12%)` | :arrow_up: | |
| [computing-unit-managing-service](https://app.codecov.io/gh/apache/texera/pull/5451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `1.65% <ø> (ø)` | | Carriedforward from [f2f0564](https://app.codecov.io/gh/apache/texera/commit/f2f056455d850f6d4f99d81ac7e128fb74906a44?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/5451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.06% <ø> (ø)` | | Carriedforward from [f2f0564](https://app.codecov.io/gh/apache/texera/commit/f2f056455d850f6d4f99d81ac7e128fb74906a44?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/5451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `38.32% <ø> (ø)` | | Carriedforward from [f2f0564](https://app.codecov.io/gh/apache/texera/commit/f2f056455d850f6d4f99d81ac7e128fb74906a44?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/5451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `46.40% <ø> (ø)` | | Carriedforward from [f2f0564](https://app.codecov.io/gh/apache/texera/commit/f2f056455d850f6d4f99d81ac7e128fb74906a44?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/5451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.69% <ø> (ø)` | | Carriedforward from [f2f0564](https://app.codecov.io/gh/apache/texera/commit/f2f056455d850f6d4f99d81ac7e128fb74906a44?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [python](https://app.codecov.io/gh/apache/texera/pull/5451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.83% <ø> (ø)` | | Carriedforward from [f2f0564](https://app.codecov.io/gh/apache/texera/commit/f2f056455d850f6d4f99d81ac7e128fb74906a44?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/5451/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.69% <ø> (ø)` | | Carriedforward from [f2f0564](https://app.codecov.io/gh/apache/texera/commit/f2f056455d850f6d4f99d81ac7e128fb74906a44?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/5451?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 unit test coverage for InputManager [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #5451:
URL: https://github.com/apache/texera/pull/5451#discussion_r3368791257
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManagerSpec.scala:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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.messaginglayer
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity}
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.Partitioning
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.DPInputQueueElement
+import org.scalatest.flatspec.AnyFlatSpec
+
+import java.net.URI
+import java.util.concurrent.LinkedBlockingQueue
+
+class InputManagerSpec extends AnyFlatSpec {
+
+ // ---------------------------------------------------------------------------
+ // Fixtures
+ // ---------------------------------------------------------------------------
+
+ private val actorId: ActorVirtualIdentity = ActorVirtualIdentity("worker-1")
+
+ private def emptyQueue(): LinkedBlockingQueue[DPInputQueueElement] =
+ new LinkedBlockingQueue[DPInputQueueElement]()
+
+ private def freshManager: InputManager = new InputManager(actorId, emptyQueue())
+
+ // A one-field schema is enough to build real Tuple instances for the batch tests.
+ private val intAttr = new Attribute("v", AttributeType.INTEGER)
+ private val schema: Schema = Schema().add(intAttr)
+ private def tuple(value: Int): Tuple =
+ Tuple.builder(schema).add(intAttr, Integer.valueOf(value)).build()
+
+ private def channelId(from: String, to: String): ChannelIdentity =
+ ChannelIdentity(ActorVirtualIdentity(from), ActorVirtualIdentity(to), isControl = false)
+
+ // ---------------------------------------------------------------------------
+ // getAllPorts / addPort / getPort
+ // ---------------------------------------------------------------------------
+
+ "InputManager.getAllPorts (fresh)" should "be empty" in {
+ val mgr = freshManager
+ assert(mgr.getAllPorts.isEmpty)
+ }
+
+ "InputManager.addPort" should
+ "register a fresh WorkerPort under the supplied portId (no reader threads when URI list is empty)" in {
+ val mgr = freshManager
+ val portId = PortIdentity(0)
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty)
+ assert(mgr.getAllPorts == Set(portId))
+ val port = mgr.getPort(portId)
+ assert(port.schema == schema)
+ assert(!port.completed)
+ assert(mgr.getInputPortReaderThreads.isEmpty)
+ }
+
+ it should "be a no-op when called a second time for the same portId (first call wins)" in {
+ // Pin the "each port can only be added and initialized once" comment in
+ // the production code: the second addPort returns early WITHOUT
+ // overwriting the existing WorkerPort. We prove this by mutating the
+ // first port's `completed` flag, then re-adding, then checking the
+ // mutation survives — if the second call had replaced the WorkerPort,
+ // `completed` would revert to its case-class default `false`.
+ val mgr = freshManager
+ val portId = PortIdentity(0)
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty)
+ val firstPort = mgr.getPort(portId)
+ firstPort.completed = true
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty)
+ assert(mgr.getPort(portId) eq firstPort, "second addPort must not replace the WorkerPort")
+ assert(mgr.getPort(portId).completed, "second addPort must not reset the existing port's state")
+ }
+
+ it should
+ "throw AssertionError when urisToRead.size does not match partitionings.size" in {
+ val mgr = freshManager
+ val portId = PortIdentity(0)
+ // 1 URI but 0 partitionings — sizes diverge.
+ assertThrows[AssertionError] {
+ mgr.addPort(
+ portId,
+ schema,
+ urisToRead = List(new URI("file:///nowhere")),
+ partitionings = List.empty[Partitioning]
+ )
+ }
+ }
+
+ "InputManager.getPort" should "return the WorkerPort previously registered by addPort" in {
+ val mgr = freshManager
+ val portId = PortIdentity(7)
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty)
+ val a = mgr.getPort(portId)
+ val b = mgr.getPort(portId)
+ assert(a eq b, "getPort must be a pure lookup — no fresh WorkerPort each call")
+ }
+
+ "InputManager.getInputPortReaderThreads" should
+ "be empty when no port has been added with URIs" in {
+ val mgr = freshManager
+ mgr.addPort(PortIdentity(0), schema, urisToRead = List.empty, partitionings = List.empty)
+ mgr.addPort(PortIdentity(1), schema, urisToRead = List.empty, partitionings = List.empty)
+ assert(mgr.getInputPortReaderThreads.isEmpty)
+ }
+
+ // ---------------------------------------------------------------------------
+ // Batch cursor — initBatch, hasUnfinishedInput, getNextTuple, getCurrentTuple
+ // ---------------------------------------------------------------------------
+
+ "InputManager.hasUnfinishedInput (fresh)" should "be false when no batch has been initialized" in {
+ val mgr = freshManager
+ assert(!mgr.hasUnfinishedInput)
+ }
+
+ "InputManager.getCurrentTuple (fresh)" should "return null when no batch has been initialized" in {
+ val mgr = freshManager
+ assert(mgr.getCurrentTuple == null)
+ }
+
+ "InputManager.initBatch" should
+ "swap in a new batch, reset the cursor, and surface the channel id" in {
Review Comment:
Fixed in 343e720855 — added a `getNextTuple` assertion that verifies the first element of the new batch is returned, directly proving `currentInputIdx` was reset.
--
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 unit test coverage for InputManager [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan merged PR #5451:
URL: https://github.com/apache/texera/pull/5451
--
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 unit test coverage for InputManager [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #5451:
URL: https://github.com/apache/texera/pull/5451#discussion_r3368764005
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManagerSpec.scala:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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.messaginglayer
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity}
+import org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings.Partitioning
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.DPInputQueueElement
+import org.scalatest.flatspec.AnyFlatSpec
+
+import java.net.URI
+import java.util.concurrent.LinkedBlockingQueue
+
+class InputManagerSpec extends AnyFlatSpec {
+
+ // ---------------------------------------------------------------------------
+ // Fixtures
+ // ---------------------------------------------------------------------------
+
+ private val actorId: ActorVirtualIdentity = ActorVirtualIdentity("worker-1")
+
+ private def emptyQueue(): LinkedBlockingQueue[DPInputQueueElement] =
+ new LinkedBlockingQueue[DPInputQueueElement]()
+
+ private def freshManager: InputManager = new InputManager(actorId, emptyQueue())
+
+ // A one-field schema is enough to build real Tuple instances for the batch tests.
+ private val intAttr = new Attribute("v", AttributeType.INTEGER)
+ private val schema: Schema = Schema().add(intAttr)
+ private def tuple(value: Int): Tuple =
+ Tuple.builder(schema).add(intAttr, Integer.valueOf(value)).build()
+
+ private def channelId(from: String, to: String): ChannelIdentity =
+ ChannelIdentity(ActorVirtualIdentity(from), ActorVirtualIdentity(to), isControl = false)
+
+ // ---------------------------------------------------------------------------
+ // getAllPorts / addPort / getPort
+ // ---------------------------------------------------------------------------
+
+ "InputManager.getAllPorts (fresh)" should "be empty" in {
+ val mgr = freshManager
+ assert(mgr.getAllPorts.isEmpty)
+ }
+
+ "InputManager.addPort" should
+ "register a fresh WorkerPort under the supplied portId (no reader threads when URI list is empty)" in {
+ val mgr = freshManager
+ val portId = PortIdentity(0)
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty)
+ assert(mgr.getAllPorts == Set(portId))
+ val port = mgr.getPort(portId)
+ assert(port.schema == schema)
+ assert(!port.completed)
+ assert(mgr.getInputPortReaderThreads.isEmpty)
+ }
+
+ it should "be a no-op when called a second time for the same portId (first call wins)" in {
+ // Pin the "each port can only be added and initialized once" comment in
+ // the production code: the second addPort returns early WITHOUT
+ // overwriting the existing WorkerPort. We prove this by mutating the
+ // first port's `completed` flag, then re-adding, then checking the
+ // mutation survives — if the second call had replaced the WorkerPort,
+ // `completed` would revert to its case-class default `false`.
+ val mgr = freshManager
+ val portId = PortIdentity(0)
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty)
+ val firstPort = mgr.getPort(portId)
+ firstPort.completed = true
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty)
+ assert(mgr.getPort(portId) eq firstPort, "second addPort must not replace the WorkerPort")
+ assert(mgr.getPort(portId).completed, "second addPort must not reset the existing port's state")
+ }
+
+ it should
+ "throw AssertionError when urisToRead.size does not match partitionings.size" in {
+ val mgr = freshManager
+ val portId = PortIdentity(0)
+ // 1 URI but 0 partitionings — sizes diverge.
+ assertThrows[AssertionError] {
+ mgr.addPort(
+ portId,
+ schema,
+ urisToRead = List(new URI("file:///nowhere")),
+ partitionings = List.empty[Partitioning]
+ )
+ }
+ }
+
+ "InputManager.getPort" should "return the WorkerPort previously registered by addPort" in {
+ val mgr = freshManager
+ val portId = PortIdentity(7)
+ mgr.addPort(portId, schema, urisToRead = List.empty, partitionings = List.empty)
+ val a = mgr.getPort(portId)
+ val b = mgr.getPort(portId)
+ assert(a eq b, "getPort must be a pure lookup — no fresh WorkerPort each call")
+ }
+
+ "InputManager.getInputPortReaderThreads" should
+ "be empty when no port has been added with URIs" in {
+ val mgr = freshManager
+ mgr.addPort(PortIdentity(0), schema, urisToRead = List.empty, partitionings = List.empty)
+ mgr.addPort(PortIdentity(1), schema, urisToRead = List.empty, partitionings = List.empty)
+ assert(mgr.getInputPortReaderThreads.isEmpty)
+ }
+
+ // ---------------------------------------------------------------------------
+ // Batch cursor — initBatch, hasUnfinishedInput, getNextTuple, getCurrentTuple
+ // ---------------------------------------------------------------------------
+
+ "InputManager.hasUnfinishedInput (fresh)" should "be false when no batch has been initialized" in {
+ val mgr = freshManager
+ assert(!mgr.hasUnfinishedInput)
+ }
+
+ "InputManager.getCurrentTuple (fresh)" should "return null when no batch has been initialized" in {
+ val mgr = freshManager
+ assert(mgr.getCurrentTuple == null)
+ }
+
+ "InputManager.initBatch" should
+ "swap in a new batch, reset the cursor, and surface the channel id" in {
Review Comment:
This test description claims it "reset[s] the cursor", but the assertions in this block only verify `currentChannelId` and `hasUnfinishedInput`. Either rename the test to match what it actually asserts, or add an assertion that demonstrates the cursor was reset (e.g., that the next tuple returned is the first element).
--
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