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/08/31 01:07:38 UTC
[PR] test(amber): cover the replay-log writer, manager and congestion control [texera]
aglinxinyuan opened a new pull request, #8177:
URL: https://github.com/apache/texera/pull/8177
### What changes were proposed in this PR?
Two new specs for classes that had none, plus one append. 33 tests across the three suites.
| File | Codecov lines | Newly covered |
|---|---|---|
| `AsyncReplayLogWriter.scala` | 38/44 → **42/44** | 51, 58, 71, 72 |
| `CongestionControl.scala` | 30/34 → **32/34** | 70, 71 |
| `ReplayLogManager.scala` | 30/34 → **31/34** | 122 |
**+7 fully-covered lines** (120/137 → 127/137 across the bundle). The plain line-hit metric moves only +3, because four of the seven were already executed and flip by completing a branch arm.
New files: `AsyncReplayLogWriterSpec` (7 tests), `ReplayLogManagerImplSpec` (6 tests), and a shared `ReplayLogWriterFixtures` support object. `AsyncReplayLogWriter` and `ReplayLogManagerImpl` had no spec of their own before this.
The lines closed are the terminated-writer arms of `assert(!stopped)` in `putLogRecords` and `putOutput`, the non-default `logInterval > 0` sleep arm, the `ssThreshold` clamp reached only after five consecutive ack timeouts, and `markAsReplayDestination`.
### Honesty about the bound
The measurement filter was 110 suite FQCNs — every spec under `amber/src/test/scala/.../engine/**` plus `clustering/**`. CI's full-module `WorkflowExecutionService/jacoco` runs 188. **Adding suites can only raise the before-figure, so +7 is an upper bound on the CI gain, not a floor.**
The specific lines closed are not plausibly reached by a `core/` operator, tuple or storage spec, so I believe +7 is also the actual figure — but a bound is what was measured and that is what is claimed.
### Four files from the original scope contribute nothing
`AmberFIFOChannel`, `WorkflowMessage`, `RecoveryPayload` and `RegionPlan` were in the bundle and are worth **zero**; they already have thorough specs, including `AmberMessageEnvelopesSpec`, whose name mirrors neither source class it covers.
**`NetworkOutputGateway`'s tests were written and then deleted.** It measured 22/25 before and after — lines 50, 51, 98 and 99 are already covered by pre-existing suites at this scope. A third jacoco run on the trimmed tree reproduces 127/137 exactly, confirming those tests contributed nothing. `NetworkOutputGatewaySpec.scala` is byte-identical to `HEAD` in this PR.
### Two hazards handled rather than discovered late
**`terminate()` can hang the build.** It does `stopped = true; writerQueue.put(TerminateSignal); gracefullyStopped.get()` — an untimed `CompletableFuture.get()` completed only at the end of `run()`. On a writer that was never started it blocks forever, and amber's suites are strictly serial in one JVM, so that hangs the module rather than failing a test. Every test starts before terminating, with `terminate()` in a `finally`; the class extends `Thread`, so a missed one would also leak a live thread into the shared JVM.
**`logInterval` is a JVM-global.** It comes from `ApplicationConfig.faultToleranceLogFlushIntervalInMs`, a memoized `val` that is 0 by default. Poking that object would poison every later suite. Instead the tests reflect on the **instance** field — `javap` confirms `private final long logInterval`, and setting it via `Field.setLong` on the instance was verified to work on this JDK. Instance-local, poisons nothing.
### Deliberately not included
`AmberFIFOChannel` lines 47/48/51 are `logger.debug` bodies. There is no `logback-test.xml`, ROOT governs, and CI sets `TEXERA_SERVICE_LOG_LEVEL=WARN`, so the scala-logging `isDebugEnabled` arm never fires in CI — they would look green locally and be missed in CI. Not counted, and not chased by raising the log level in a test, which would mutate a shared logging context in a serial JVM.
`NetworkOutputGateway:41` is the `AmberLogging` lazy `logger` accessor, and the class body contains no logging call at all, so there is no call site to reach it from. Covering it needs a production seam — refused.
### Verification
All mutations in the final table die, including the four the reviewers measured as surviving the first draft (complete-before-close, flush-after-release, drop-`Left`s, and the `INIT_STEP` mutant), re-verified on the final tree.
Four `NetworkOutputGateway` mutants are recorded as killed by **pre-existing** suites without help from this bundle, rather than credited to it.
Measured with one fresh sbt JVM per side, the jacoco directory removed between runs, an identical throwaway suite-name filter both times (deleted afterwards), and `STORAGE_ICEBERG_CATALOG_TYPE=postgres` with per-worktree namespaces — which takes the local baseline from 28 failures and 1 abort to **zero**, so neither run is a partial or all-zero report.
No production file is touched.
### Any related issues, documentation, discussions?
Closes #8175
### How was this PR tested?
```
sbt "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.logreplay.AsyncReplayLogWriterSpec org.apache.texera.amber.engine.architecture.logreplay.ReplayLogManagerImplSpec org.apache.texera.amber.engine.architecture.messaginglayer.CongestionControlSpec"
```
```
[info] Total number of tests run: 33
[info] Tests: succeeded 33, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
```
`WorkflowExecutionService/Test/scalafmtCheck` and `WorkflowExecutionService/Test/scalafix --check` both pass. Re-run after rebasing onto current `main`.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
--
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): cover the replay-log writer, manager and congestion control [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #8177:
URL: https://github.com/apache/texera/pull/8177#discussion_r3896207285
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/CongestionControlSpec.scala:
##########
@@ -174,6 +174,34 @@ class CongestionControlSpec extends AnyFlatSpec {
)
}
+ it should "clamp ssThreshold at 1 so repeated timeouts never collapse the window to zero" in {
+ // Five consecutive timed-out acks halve ssThreshold 16→8→4→2→1→0. The clamp
+ // must lift that final 0 back to 1, because windowSize is then set from
+ // ssThreshold and `canSend` is `inTransit.size < windowSize`: a window of 0
+ // can never be satisfied, so the sender would wedge permanently.
+ val cc = new CongestionControl()
+ (1L to 5L).foreach { i =>
+ cc.markMessageInTransit(msg(i))
+ backdateSentTime(cc, i, 5000) // > ackTimeLimit (3000)
+ cc.ack(i)
+ }
+ assert(
+ cc.getStatusReport == "current window size = 1 \t in transit = 0 \t waiting = 0",
+ s"unexpected status: ${cc.getStatusReport}"
+ )
Review Comment:
Premise accepted, remedy adjusted — `be53e5c4` and `1ce599bd`. The clamp test no longer compares the `getStatusReport` string at all; it asserts the `canSend` boundary directly (`getInTransitMessages.isEmpty` and `canSend`, then `markMessageInTransit` and `!canSend`, i.e. `windowSize == 1` exactly, twice). `getStatusReport` is still built, but as the failure clue rather than the assertion.
Re-verifying this turned up something the comment did not name: the *dedicated* format test drove all three counters to `1` — `current window size = 1 \t in transit = 1 \t waiting = 1`. A string where every field holds the same value cannot distinguish the fields, so a mutant transposing two of them survived. The counters are now driven to three distinct values (`2 / 1 / 3`), which is what the underlying concern was really pointing at.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogWriterFixtures.scala:
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.logreplay
+
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.MainThreadDelegateMessage
+import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage
+import org.apache.texera.amber.engine.common.storage.SequentialRecordStorage.SequentialRecordWriter
+
+import java.io.{ByteArrayOutputStream, DataOutputStream}
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.{ConcurrentLinkedQueue, CountDownLatch, TimeUnit}
+import scala.jdk.CollectionConverters.CollectionHasAsScala
+
+/**
+ * In-process fixtures shared by `AsyncReplayLogWriterSpec` and
+ * `ReplayLogManagerImplSpec`, the two specs that drive a real
+ * `AsyncReplayLogWriter` thread. Nothing here touches the filesystem, an
+ * ActorSystem, or `AmberRuntime.serde`.
+ */
+object ReplayLogWriterFixtures {
+
+ /** Ordered trace of everything the writer thread did, in the order it did it. */
+ sealed trait WriterEvent
+ final case class Wrote(record: ReplayLogRecord) extends WriterEvent
+ final case class Sent(msg: Either[MainThreadDelegateMessage, WorkflowFIFOMessage])
+ extends WriterEvent
+
+ /**
+ * A flush that actually made previously written records durable. Without this
+ * event a trace cannot tell `write; flush; send` from `write; send; flush`,
+ * and the second is a durability inversion: the message is on the network
+ * while its log record is still in the `DataOutputStream` buffer.
+ */
+ case object Flushed extends WriterEvent
+ case object Closed extends WriterEvent
+
+ /**
+ * Thread-safe recorder. The writer thread appends; the test thread reads.
+ * `ConcurrentLinkedQueue` supplies both FIFO ordering and the happens-before
+ * edge, so no extra synchronisation is needed on the assertion side.
+ */
+ class Recorder {
+ private val events = new ConcurrentLinkedQueue[WriterEvent]()
+ private val firstEvent = new CountDownLatch(1)
+
+ def record(event: WriterEvent): Unit = {
+ events.add(event)
+ firstEvent.countDown()
+ }
+
+ /** Blocks until the writer thread produces its first event, or the timeout elapses. */
+ def awaitFirstEvent(timeoutMillis: Long): Boolean =
+ firstEvent.await(timeoutMillis, TimeUnit.MILLISECONDS)
+
+ def snapshot: List[WriterEvent] = events.asScala.toList
+
+ /** The `handler` an AsyncReplayLogWriter calls for each released output. */
+ val handler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit =
+ msg => record(Sent(msg))
+ }
+
+ /**
+ * A `SequentialRecordWriter` that records rather than serialises. Overriding
+ * `writeRecord` keeps `AmberRuntime.serde` (and therefore a Pekko
+ * `ActorSystem`) out of these specs entirely; the wrapped stream is never
+ * written to and the base class's `lazy val output` is never forced.
+ *
+ * `close()` sleeps before recording. Under pristine code that costs nothing
+ * in correctness terms — `run()` closes the writer and only then completes
+ * the future `terminate()` waits on, so the `Closed` event is ordered before
+ * `terminate()` can return no matter how long the close takes. It is what
+ * turns "did close happen before terminate returned?" from a nanosecond race
+ * into a decided question: an implementation that completed the future first
+ * would hand the assertion a trace with no `Closed` in it.
+ *
+ * `flush()` records only when something has been written since the last
+ * flush. The real writer's `flush()` is `Output.flush()`, a no-op on an empty
+ * buffer, so a flush with nothing buffered changes no durability and is not
+ * worth a trace entry — and leaving it out keeps the trace independent of how
+ * the writer thread happens to split its drain batches.
+ */
+ class RecordingRecordWriter(recorder: Recorder)
+ extends SequentialRecordWriter[ReplayLogRecord](
+ new DataOutputStream(new ByteArrayOutputStream())
+ ) {
+ // Touched only by the writer thread (writeRecord/flush/close all run there).
+ private var buffered = false
+
+ override def writeRecord(obj: ReplayLogRecord): Unit = {
+ buffered = true
+ recorder.record(Wrote(obj))
+ }
+
+ override def flush(): Unit =
+ if (buffered) {
+ buffered = false
+ recorder.record(Flushed)
+ }
+
+ override def close(): Unit = {
+ Thread.sleep(CloseDelayMillis)
+ recorder.record(Closed)
+ }
+ }
+
+ /** See `RecordingRecordWriter.close()`. */
+ val CloseDelayMillis = 250L
+
+ /**
+ * `AsyncReplayLogWriter.terminate()` blocks on an untimed
+ * `CompletableFuture.get()` that is only completed at the very end of
+ * `run()`. amber suites are strictly serial (`Tags.limit(Tags.Test, 1)`), so
+ * a writer that never finishes would hang the whole module build rather than
+ * fail one test. Run the shutdown on a daemon thread and bound the wait, so a
+ * wedge surfaces as an ordinary failed assertion instead.
+ *
+ * The trace is captured on the shutdown thread the instant `shutdown()`
+ * returns, so assertions see exactly what the writer had done by the time the
+ * caller was released — not whatever it managed to do afterwards while the
+ * test thread was getting around to reading.
+ *
+ * @return (whether the shutdown returned inside the budget, the trace as of
+ * the moment it returned).
+ */
+ def terminatesWithin(timeoutMillis: Long, recorder: Recorder)(
+ shutdown: () => Unit
+ ): (Boolean, List[WriterEvent]) = {
+ val captured = new AtomicReference[List[WriterEvent]](Nil)
+ val t = new Thread(() => {
+ try {
+ shutdown()
+ captured.set(recorder.snapshot)
+ } catch { case _: Throwable => () }
+ })
+ t.setDaemon(true)
+ t.start()
+ t.join(timeoutMillis)
Review Comment:
Applied in `be53e5c4`. `terminatesWithin` now captures what `shutdown()` threw in an `AtomicReference` and rethrows the original instance on the calling thread once `join()` has returned, so the cause surfaces instead of being discarded. The thread local was renamed `t` → `shutdownThread` so the catch binding no longer shadows it.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogWriterFixtures.scala:
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.logreplay
+
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.MainThreadDelegateMessage
+import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage
+import org.apache.texera.amber.engine.common.storage.SequentialRecordStorage.SequentialRecordWriter
+
+import java.io.{ByteArrayOutputStream, DataOutputStream}
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.{ConcurrentLinkedQueue, CountDownLatch, TimeUnit}
+import scala.jdk.CollectionConverters.CollectionHasAsScala
+
+/**
+ * In-process fixtures shared by `AsyncReplayLogWriterSpec` and
+ * `ReplayLogManagerImplSpec`, the two specs that drive a real
+ * `AsyncReplayLogWriter` thread. Nothing here touches the filesystem, an
+ * ActorSystem, or `AmberRuntime.serde`.
+ */
+object ReplayLogWriterFixtures {
+
+ /** Ordered trace of everything the writer thread did, in the order it did it. */
+ sealed trait WriterEvent
+ final case class Wrote(record: ReplayLogRecord) extends WriterEvent
+ final case class Sent(msg: Either[MainThreadDelegateMessage, WorkflowFIFOMessage])
+ extends WriterEvent
+
+ /**
+ * A flush that actually made previously written records durable. Without this
+ * event a trace cannot tell `write; flush; send` from `write; send; flush`,
+ * and the second is a durability inversion: the message is on the network
+ * while its log record is still in the `DataOutputStream` buffer.
+ */
+ case object Flushed extends WriterEvent
+ case object Closed extends WriterEvent
+
+ /**
+ * Thread-safe recorder. The writer thread appends; the test thread reads.
+ * `ConcurrentLinkedQueue` supplies both FIFO ordering and the happens-before
+ * edge, so no extra synchronisation is needed on the assertion side.
+ */
+ class Recorder {
+ private val events = new ConcurrentLinkedQueue[WriterEvent]()
+ private val firstEvent = new CountDownLatch(1)
+
+ def record(event: WriterEvent): Unit = {
+ events.add(event)
+ firstEvent.countDown()
+ }
+
+ /** Blocks until the writer thread produces its first event, or the timeout elapses. */
+ def awaitFirstEvent(timeoutMillis: Long): Boolean =
+ firstEvent.await(timeoutMillis, TimeUnit.MILLISECONDS)
+
+ def snapshot: List[WriterEvent] = events.asScala.toList
+
+ /** The `handler` an AsyncReplayLogWriter calls for each released output. */
+ val handler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit =
+ msg => record(Sent(msg))
+ }
+
+ /**
+ * A `SequentialRecordWriter` that records rather than serialises. Overriding
+ * `writeRecord` keeps `AmberRuntime.serde` (and therefore a Pekko
+ * `ActorSystem`) out of these specs entirely; the wrapped stream is never
+ * written to and the base class's `lazy val output` is never forced.
+ *
+ * `close()` sleeps before recording. Under pristine code that costs nothing
+ * in correctness terms — `run()` closes the writer and only then completes
+ * the future `terminate()` waits on, so the `Closed` event is ordered before
+ * `terminate()` can return no matter how long the close takes. It is what
+ * turns "did close happen before terminate returned?" from a nanosecond race
+ * into a decided question: an implementation that completed the future first
+ * would hand the assertion a trace with no `Closed` in it.
+ *
+ * `flush()` records only when something has been written since the last
+ * flush. The real writer's `flush()` is `Output.flush()`, a no-op on an empty
+ * buffer, so a flush with nothing buffered changes no durability and is not
+ * worth a trace entry — and leaving it out keeps the trace independent of how
+ * the writer thread happens to split its drain batches.
+ */
+ class RecordingRecordWriter(recorder: Recorder)
+ extends SequentialRecordWriter[ReplayLogRecord](
+ new DataOutputStream(new ByteArrayOutputStream())
+ ) {
+ // Touched only by the writer thread (writeRecord/flush/close all run there).
+ private var buffered = false
+
+ override def writeRecord(obj: ReplayLogRecord): Unit = {
+ buffered = true
+ recorder.record(Wrote(obj))
+ }
+
+ override def flush(): Unit =
+ if (buffered) {
+ buffered = false
+ recorder.record(Flushed)
+ }
+
+ override def close(): Unit = {
+ Thread.sleep(CloseDelayMillis)
+ recorder.record(Closed)
+ }
Review Comment:
Refused. That `Thread.sleep` is not incidental latency — it is the device that makes the durability ordering decidable, and it is load-bearing for more than one test.
`AsyncReplayLogWriter.run()` closes the record writer and only *then* completes the future `terminate()` waits on, and the fixture snapshots the trace on the shutdown thread the instant `terminate()` returns. Without a delay inside `close()`, "did the writer finish flushing before it signalled completion?" stops being an observable question and the ordering assertions become vacuous rather than fast. A latch would let the fixture *observe* the ordering but not *require* it.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogWriterFixtures.scala:
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.logreplay
+
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.MainThreadDelegateMessage
+import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage
+import org.apache.texera.amber.engine.common.storage.SequentialRecordStorage.SequentialRecordWriter
+
+import java.io.{ByteArrayOutputStream, DataOutputStream}
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.{ConcurrentLinkedQueue, CountDownLatch, TimeUnit}
+import scala.jdk.CollectionConverters.CollectionHasAsScala
+
+/**
+ * In-process fixtures shared by `AsyncReplayLogWriterSpec` and
+ * `ReplayLogManagerImplSpec`, the two specs that drive a real
+ * `AsyncReplayLogWriter` thread. Nothing here touches the filesystem, an
+ * ActorSystem, or `AmberRuntime.serde`.
+ */
+object ReplayLogWriterFixtures {
+
+ /** Ordered trace of everything the writer thread did, in the order it did it. */
+ sealed trait WriterEvent
+ final case class Wrote(record: ReplayLogRecord) extends WriterEvent
+ final case class Sent(msg: Either[MainThreadDelegateMessage, WorkflowFIFOMessage])
+ extends WriterEvent
+
+ /**
+ * A flush that actually made previously written records durable. Without this
+ * event a trace cannot tell `write; flush; send` from `write; send; flush`,
+ * and the second is a durability inversion: the message is on the network
+ * while its log record is still in the `DataOutputStream` buffer.
+ */
+ case object Flushed extends WriterEvent
+ case object Closed extends WriterEvent
+
+ /**
+ * Thread-safe recorder. The writer thread appends; the test thread reads.
+ * `ConcurrentLinkedQueue` supplies both FIFO ordering and the happens-before
+ * edge, so no extra synchronisation is needed on the assertion side.
+ */
+ class Recorder {
+ private val events = new ConcurrentLinkedQueue[WriterEvent]()
+ private val firstEvent = new CountDownLatch(1)
+
+ def record(event: WriterEvent): Unit = {
+ events.add(event)
+ firstEvent.countDown()
+ }
+
+ /** Blocks until the writer thread produces its first event, or the timeout elapses. */
+ def awaitFirstEvent(timeoutMillis: Long): Boolean =
+ firstEvent.await(timeoutMillis, TimeUnit.MILLISECONDS)
+
+ def snapshot: List[WriterEvent] = events.asScala.toList
+
+ /** The `handler` an AsyncReplayLogWriter calls for each released output. */
+ val handler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit =
+ msg => record(Sent(msg))
+ }
+
+ /**
+ * A `SequentialRecordWriter` that records rather than serialises. Overriding
+ * `writeRecord` keeps `AmberRuntime.serde` (and therefore a Pekko
+ * `ActorSystem`) out of these specs entirely; the wrapped stream is never
+ * written to and the base class's `lazy val output` is never forced.
+ *
+ * `close()` sleeps before recording. Under pristine code that costs nothing
+ * in correctness terms — `run()` closes the writer and only then completes
+ * the future `terminate()` waits on, so the `Closed` event is ordered before
+ * `terminate()` can return no matter how long the close takes. It is what
+ * turns "did close happen before terminate returned?" from a nanosecond race
+ * into a decided question: an implementation that completed the future first
+ * would hand the assertion a trace with no `Closed` in it.
+ *
+ * `flush()` records only when something has been written since the last
+ * flush. The real writer's `flush()` is `Output.flush()`, a no-op on an empty
+ * buffer, so a flush with nothing buffered changes no durability and is not
+ * worth a trace entry — and leaving it out keeps the trace independent of how
+ * the writer thread happens to split its drain batches.
+ */
+ class RecordingRecordWriter(recorder: Recorder)
+ extends SequentialRecordWriter[ReplayLogRecord](
+ new DataOutputStream(new ByteArrayOutputStream())
+ ) {
+ // Touched only by the writer thread (writeRecord/flush/close all run there).
+ private var buffered = false
+
+ override def writeRecord(obj: ReplayLogRecord): Unit = {
+ buffered = true
+ recorder.record(Wrote(obj))
+ }
+
+ override def flush(): Unit =
+ if (buffered) {
+ buffered = false
+ recorder.record(Flushed)
+ }
+
+ override def close(): Unit = {
+ Thread.sleep(CloseDelayMillis)
+ recorder.record(Closed)
+ }
+ }
+
+ /** See `RecordingRecordWriter.close()`. */
+ val CloseDelayMillis = 250L
+
+ /**
+ * `AsyncReplayLogWriter.terminate()` blocks on an untimed
+ * `CompletableFuture.get()` that is only completed at the very end of
+ * `run()`. amber suites are strictly serial (`Tags.limit(Tags.Test, 1)`), so
+ * a writer that never finishes would hang the whole module build rather than
+ * fail one test. Run the shutdown on a daemon thread and bound the wait, so a
+ * wedge surfaces as an ordinary failed assertion instead.
+ *
+ * The trace is captured on the shutdown thread the instant `shutdown()`
+ * returns, so assertions see exactly what the writer had done by the time the
+ * caller was released — not whatever it managed to do afterwards while the
+ * test thread was getting around to reading.
+ *
+ * @return (whether the shutdown returned inside the budget, the trace as of
+ * the moment it returned).
+ */
+ def terminatesWithin(timeoutMillis: Long, recorder: Recorder)(
+ shutdown: () => Unit
+ ): (Boolean, List[WriterEvent]) = {
+ val captured = new AtomicReference[List[WriterEvent]](Nil)
+ val t = new Thread(() => {
+ try {
+ shutdown()
+ captured.set(recorder.snapshot)
+ } catch { case _: Throwable => () }
+ })
+ t.setDaemon(true)
+ t.start()
+ t.join(timeoutMillis)
+ (!t.isAlive, captured.get())
Review Comment:
Applied in `be53e5c4`. After `join(timeoutMillis)` the helper computes `returned = !shutdownThread.isAlive`, calls `interrupt()` when the budget elapsed, and only consults the captured throwable when the thread actually returned — so a wedged shutdown no longer leaves a live thread interacting with shared state.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/AsyncReplayLogWriterSpec.scala:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.logreplay
+
+import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity}
+import org.apache.texera.amber.engine.architecture.logreplay.ReplayLogWriterFixtures.{
+ Closed,
+ Flushed,
+ Recorder,
+ RecordingRecordWriter,
+ Sent,
+ WriterEvent,
+ Wrote,
+ terminatesWithin
+}
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.MainThreadDelegateMessage
+import org.apache.texera.amber.engine.common.ambermessage.{DataFrame, WorkflowFIFOMessage}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class AsyncReplayLogWriterSpec extends AnyFlatSpec {
+
+ private val channel =
+ ChannelIdentity(ActorVirtualIdentity("from"), ActorVirtualIdentity("to"), isControl = false)
+
+ private def fifo(seq: Long): WorkflowFIFOMessage =
+ WorkflowFIFOMessage(channel, seq, DataFrame(Array.empty))
+
+ private def output(seq: Long): Either[MainThreadDelegateMessage, WorkflowFIFOMessage] =
+ Right(fifo(seq))
+
+ /**
+ * The other arm of the queue element's `Either`. `DPThread`,
+ * `PrepareCheckpointHandler` and `FinalizeCheckpointHandler` all push
+ * `Left(MainThreadDelegateMessage(...))` through this writer, so dropping one
+ * would break checkpointing silently rather than loudly.
+ */
+ private def delegate(): Either[MainThreadDelegateMessage, WorkflowFIFOMessage] =
+ Left(MainThreadDelegateMessage(_ => ()))
+
+ private def newWriter(recorder: Recorder): AsyncReplayLogWriter = {
+ val writer = new AsyncReplayLogWriter(recorder.handler, new RecordingRecordWriter(recorder))
+ // Daemon so that even a wedged writer can never keep a JVM from exiting.
+ writer.setDaemon(true)
+ writer
+ }
+
+ private val shutdownBudgetMillis = 30000L
+
+ /**
+ * Shuts the writer down and returns the trace as of the instant
+ * `terminate()` returned. Bounded, because `terminate()` blocks on an untimed
+ * future (see `ReplayLogWriterFixtures.terminatesWithin`).
+ */
+ private def shutdownTrace(
+ writer: AsyncReplayLogWriter,
+ recorder: Recorder
+ ): List[WriterEvent] = {
+ val (returned, trace) =
+ terminatesWithin(shutdownBudgetMillis, recorder)(() => writer.terminate())
+ assert(returned, "terminate() did not return")
+ trace
+ }
+
+ /**
+ * `logInterval` is copied from `ApplicationConfig.faultToleranceLogFlushIntervalInMs`,
+ * a memoised `val` on a shared object that must NOT be poked: amber suites run
+ * serially in one JVM, so mutating it would poison every later suite. The writer
+ * stores its own copy in a `private final long` *instance* field, and overwriting
+ * that is instance-local. Done before `start()`, so `Thread.start()`'s
+ * happens-before edge publishes the new value to the writer thread.
+ */
+ private def setLogInterval(writer: AsyncReplayLogWriter, millis: Long): Unit = {
+ val field = classOf[AsyncReplayLogWriter].getDeclaredField("logInterval")
+ field.setAccessible(true)
+ field.setLong(writer, millis)
+ assert(field.getLong(writer) == millis, "reflective logInterval override did not take effect")
+ }
+
+ /** A writer that has completed its full start/terminate lifecycle. */
+ private def startedAndTerminated(): AsyncReplayLogWriter = {
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ writer.start()
+ shutdownTrace(writer, recorder)
+ writer
+ }
+
+ "AsyncReplayLogWriter" should
+ "write and flush queued log records before releasing the queued outputs" in {
+ // The class exists to guarantee this ordering: a message must not reach the
+ // network before the log record that would let a replay reproduce it — and
+ // "written" here has to mean flushed, not merely handed to the stream. Both
+ // items are queued *before* start(), so the writer thread's first drainTo is
+ // guaranteed to pick up both in one batch, which is exactly the batch in
+ // which the ordering can be got wrong.
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ val step = ProcessingStep(channel, 0L)
+ val released = output(1L)
+
+ writer.putLogRecords(Array(step))
+ writer.putOutput(released)
+ writer.start()
+
+ assert(shutdownTrace(writer, recorder) == List(Wrote(step), Flushed, Sent(released), Closed))
+ }
+
+ it should "write every queued record, in queue order, before the outputs that follow them" in {
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ val first = ProcessingStep(channel, 0L)
+ val second = MessageContent(fifo(7L))
+ val released = output(2L)
+
+ writer.putLogRecords(Array(first, second))
+ writer.putOutput(released)
+ writer.start()
+
+ assert(
+ shutdownTrace(writer, recorder) ==
+ List(Wrote(first), Wrote(second), Flushed, Sent(released), Closed)
+ )
+ }
+
+ it should "release main-thread delegates as well as FIFO messages, in queue order" in {
+ // The queue element is an Either precisely so that main-thread delegates
+ // (checkpoint closures) ride the same ordering guarantee as network
+ // messages. A writer that released only the Right arm would drop every
+ // checkpoint closure while still writing its log records.
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ val step = ProcessingStep(channel, 0L)
+ val closure = delegate()
+ val released = output(3L)
+
+ writer.putLogRecords(Array(step))
+ writer.putOutput(closure)
+ writer.putOutput(released)
+ writer.start()
+
+ assert(
+ shutdownTrace(writer, recorder) ==
+ List(Wrote(step), Flushed, Sent(closure), Sent(released), Closed)
+ )
+ }
+
+ it should "close the underlying record writer exactly once when it shuts down" in {
+ // terminate() is the only shutdown path, and it must both stop the thread
+ // and close the record writer — a writer left open would leak the log file
+ // handle for the rest of the worker's life. Asserting on the trace captured
+ // the instant terminate() returned is what makes this an ordering claim
+ // rather than a race: an implementation that released the caller before
+ // closing is caught even though it does eventually close.
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ writer.start()
+
+ assert(shutdownTrace(writer, recorder) == List(Closed))
+ }
+
+ it should "reject further log records once it has been terminated" in {
+ val writer = startedAndTerminated()
+ intercept[AssertionError] {
+ writer.putLogRecords(Array(ProcessingStep(channel, 0L)))
+ }
+ }
+
+ it should "reject further outputs once it has been terminated" in {
+ val writer = startedAndTerminated()
+ intercept[AssertionError] {
+ writer.putOutput(output(1L))
+ }
+ }
+
+ it should "wait for the configured flush interval before draining the queue" in {
+ // With a positive faultToleranceLogFlushIntervalInMs the writer batches by
+ // sleeping at the top of every drain loop. The default is 0 (no sleep), so
+ // this arm is only reachable with a non-default interval.
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ val flushIntervalMillis = 300L
+ setLogInterval(writer, flushIntervalMillis)
+ val step = ProcessingStep(channel, 0L)
+ writer.putLogRecords(Array(step))
+
+ val startedAt = System.nanoTime()
+ writer.start()
+ assert(recorder.awaitFirstEvent(shutdownBudgetMillis), "the queued record was never written")
+ val elapsedMillis = (System.nanoTime() - startedAt) / 1000000L
+
+ assert(shutdownTrace(writer, recorder) == List(Wrote(step), Flushed, Closed))
+ assert(
+ elapsedMillis >= 250L,
+ s"the first flush landed after only ${elapsedMillis}ms, so the ${flushIntervalMillis}ms " +
+ "flush interval was not honoured"
+ )
Review Comment:
Applied in `be53e5c4`. Added a named `sleepGranularityToleranceMillis = 50L` and replaced the literal with `flushIntervalMillis - sleepGranularityToleranceMillis`, so the bound tracks the interval it is validating. The failure message now names both the floor and the interval it came from.
--
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): cover the replay-log writer, manager and congestion control [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #8177:
URL: https://github.com/apache/texera/pull/8177#discussion_r3893942530
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/CongestionControlSpec.scala:
##########
@@ -174,6 +174,34 @@ class CongestionControlSpec extends AnyFlatSpec {
)
}
+ it should "clamp ssThreshold at 1 so repeated timeouts never collapse the window to zero" in {
+ // Five consecutive timed-out acks halve ssThreshold 16→8→4→2→1→0. The clamp
+ // must lift that final 0 back to 1, because windowSize is then set from
+ // ssThreshold and `canSend` is `inTransit.size < windowSize`: a window of 0
+ // can never be satisfied, so the sender would wedge permanently.
+ val cc = new CongestionControl()
+ (1L to 5L).foreach { i =>
+ cc.markMessageInTransit(msg(i))
+ backdateSentTime(cc, i, 5000) // > ackTimeLimit (3000)
+ cc.ack(i)
+ }
+ assert(
+ cc.getStatusReport == "current window size = 1 \t in transit = 0 \t waiting = 0",
+ s"unexpected status: ${cc.getStatusReport}"
+ )
Review Comment:
This assertion is tightly coupled to the exact formatting of `getStatusReport` (whitespace, ordering, wording). That makes the test fragile to harmless formatting changes and can cause noisy failures. Prefer asserting on stable semantics (e.g., window size / in-transit / waiting values) by parsing out the numbers or checking for key substrings rather than full-string equality.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogWriterFixtures.scala:
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.logreplay
+
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.MainThreadDelegateMessage
+import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage
+import org.apache.texera.amber.engine.common.storage.SequentialRecordStorage.SequentialRecordWriter
+
+import java.io.{ByteArrayOutputStream, DataOutputStream}
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.{ConcurrentLinkedQueue, CountDownLatch, TimeUnit}
+import scala.jdk.CollectionConverters.CollectionHasAsScala
+
+/**
+ * In-process fixtures shared by `AsyncReplayLogWriterSpec` and
+ * `ReplayLogManagerImplSpec`, the two specs that drive a real
+ * `AsyncReplayLogWriter` thread. Nothing here touches the filesystem, an
+ * ActorSystem, or `AmberRuntime.serde`.
+ */
+object ReplayLogWriterFixtures {
+
+ /** Ordered trace of everything the writer thread did, in the order it did it. */
+ sealed trait WriterEvent
+ final case class Wrote(record: ReplayLogRecord) extends WriterEvent
+ final case class Sent(msg: Either[MainThreadDelegateMessage, WorkflowFIFOMessage])
+ extends WriterEvent
+
+ /**
+ * A flush that actually made previously written records durable. Without this
+ * event a trace cannot tell `write; flush; send` from `write; send; flush`,
+ * and the second is a durability inversion: the message is on the network
+ * while its log record is still in the `DataOutputStream` buffer.
+ */
+ case object Flushed extends WriterEvent
+ case object Closed extends WriterEvent
+
+ /**
+ * Thread-safe recorder. The writer thread appends; the test thread reads.
+ * `ConcurrentLinkedQueue` supplies both FIFO ordering and the happens-before
+ * edge, so no extra synchronisation is needed on the assertion side.
+ */
+ class Recorder {
+ private val events = new ConcurrentLinkedQueue[WriterEvent]()
+ private val firstEvent = new CountDownLatch(1)
+
+ def record(event: WriterEvent): Unit = {
+ events.add(event)
+ firstEvent.countDown()
+ }
+
+ /** Blocks until the writer thread produces its first event, or the timeout elapses. */
+ def awaitFirstEvent(timeoutMillis: Long): Boolean =
+ firstEvent.await(timeoutMillis, TimeUnit.MILLISECONDS)
+
+ def snapshot: List[WriterEvent] = events.asScala.toList
+
+ /** The `handler` an AsyncReplayLogWriter calls for each released output. */
+ val handler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit =
+ msg => record(Sent(msg))
+ }
+
+ /**
+ * A `SequentialRecordWriter` that records rather than serialises. Overriding
+ * `writeRecord` keeps `AmberRuntime.serde` (and therefore a Pekko
+ * `ActorSystem`) out of these specs entirely; the wrapped stream is never
+ * written to and the base class's `lazy val output` is never forced.
+ *
+ * `close()` sleeps before recording. Under pristine code that costs nothing
+ * in correctness terms — `run()` closes the writer and only then completes
+ * the future `terminate()` waits on, so the `Closed` event is ordered before
+ * `terminate()` can return no matter how long the close takes. It is what
+ * turns "did close happen before terminate returned?" from a nanosecond race
+ * into a decided question: an implementation that completed the future first
+ * would hand the assertion a trace with no `Closed` in it.
+ *
+ * `flush()` records only when something has been written since the last
+ * flush. The real writer's `flush()` is `Output.flush()`, a no-op on an empty
+ * buffer, so a flush with nothing buffered changes no durability and is not
+ * worth a trace entry — and leaving it out keeps the trace independent of how
+ * the writer thread happens to split its drain batches.
+ */
+ class RecordingRecordWriter(recorder: Recorder)
+ extends SequentialRecordWriter[ReplayLogRecord](
+ new DataOutputStream(new ByteArrayOutputStream())
+ ) {
+ // Touched only by the writer thread (writeRecord/flush/close all run there).
+ private var buffered = false
+
+ override def writeRecord(obj: ReplayLogRecord): Unit = {
+ buffered = true
+ recorder.record(Wrote(obj))
+ }
+
+ override def flush(): Unit =
+ if (buffered) {
+ buffered = false
+ recorder.record(Flushed)
+ }
+
+ override def close(): Unit = {
+ Thread.sleep(CloseDelayMillis)
+ recorder.record(Closed)
+ }
+ }
+
+ /** See `RecordingRecordWriter.close()`. */
+ val CloseDelayMillis = 250L
+
+ /**
+ * `AsyncReplayLogWriter.terminate()` blocks on an untimed
+ * `CompletableFuture.get()` that is only completed at the very end of
+ * `run()`. amber suites are strictly serial (`Tags.limit(Tags.Test, 1)`), so
+ * a writer that never finishes would hang the whole module build rather than
+ * fail one test. Run the shutdown on a daemon thread and bound the wait, so a
+ * wedge surfaces as an ordinary failed assertion instead.
+ *
+ * The trace is captured on the shutdown thread the instant `shutdown()`
+ * returns, so assertions see exactly what the writer had done by the time the
+ * caller was released — not whatever it managed to do afterwards while the
+ * test thread was getting around to reading.
+ *
+ * @return (whether the shutdown returned inside the budget, the trace as of
+ * the moment it returned).
+ */
+ def terminatesWithin(timeoutMillis: Long, recorder: Recorder)(
+ shutdown: () => Unit
+ ): (Boolean, List[WriterEvent]) = {
+ val captured = new AtomicReference[List[WriterEvent]](Nil)
+ val t = new Thread(() => {
+ try {
+ shutdown()
+ captured.set(recorder.snapshot)
+ } catch { case _: Throwable => () }
+ })
+ t.setDaemon(true)
+ t.start()
+ t.join(timeoutMillis)
Review Comment:
Catching and discarding all `Throwable` here can hide the real cause of test failures (e.g., `InterruptedException`, assertion failures inside shutdown, or unexpected runtime errors), and it also prevents surfacing useful diagnostics. Consider capturing the thrown exception (e.g., in an `AtomicReference[Throwable]`) and rethrowing/failing on the calling thread after `join`, so failures point to the correct root cause.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogWriterFixtures.scala:
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.logreplay
+
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.MainThreadDelegateMessage
+import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage
+import org.apache.texera.amber.engine.common.storage.SequentialRecordStorage.SequentialRecordWriter
+
+import java.io.{ByteArrayOutputStream, DataOutputStream}
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.{ConcurrentLinkedQueue, CountDownLatch, TimeUnit}
+import scala.jdk.CollectionConverters.CollectionHasAsScala
+
+/**
+ * In-process fixtures shared by `AsyncReplayLogWriterSpec` and
+ * `ReplayLogManagerImplSpec`, the two specs that drive a real
+ * `AsyncReplayLogWriter` thread. Nothing here touches the filesystem, an
+ * ActorSystem, or `AmberRuntime.serde`.
+ */
+object ReplayLogWriterFixtures {
+
+ /** Ordered trace of everything the writer thread did, in the order it did it. */
+ sealed trait WriterEvent
+ final case class Wrote(record: ReplayLogRecord) extends WriterEvent
+ final case class Sent(msg: Either[MainThreadDelegateMessage, WorkflowFIFOMessage])
+ extends WriterEvent
+
+ /**
+ * A flush that actually made previously written records durable. Without this
+ * event a trace cannot tell `write; flush; send` from `write; send; flush`,
+ * and the second is a durability inversion: the message is on the network
+ * while its log record is still in the `DataOutputStream` buffer.
+ */
+ case object Flushed extends WriterEvent
+ case object Closed extends WriterEvent
+
+ /**
+ * Thread-safe recorder. The writer thread appends; the test thread reads.
+ * `ConcurrentLinkedQueue` supplies both FIFO ordering and the happens-before
+ * edge, so no extra synchronisation is needed on the assertion side.
+ */
+ class Recorder {
+ private val events = new ConcurrentLinkedQueue[WriterEvent]()
+ private val firstEvent = new CountDownLatch(1)
+
+ def record(event: WriterEvent): Unit = {
+ events.add(event)
+ firstEvent.countDown()
+ }
+
+ /** Blocks until the writer thread produces its first event, or the timeout elapses. */
+ def awaitFirstEvent(timeoutMillis: Long): Boolean =
+ firstEvent.await(timeoutMillis, TimeUnit.MILLISECONDS)
+
+ def snapshot: List[WriterEvent] = events.asScala.toList
+
+ /** The `handler` an AsyncReplayLogWriter calls for each released output. */
+ val handler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit =
+ msg => record(Sent(msg))
+ }
+
+ /**
+ * A `SequentialRecordWriter` that records rather than serialises. Overriding
+ * `writeRecord` keeps `AmberRuntime.serde` (and therefore a Pekko
+ * `ActorSystem`) out of these specs entirely; the wrapped stream is never
+ * written to and the base class's `lazy val output` is never forced.
+ *
+ * `close()` sleeps before recording. Under pristine code that costs nothing
+ * in correctness terms — `run()` closes the writer and only then completes
+ * the future `terminate()` waits on, so the `Closed` event is ordered before
+ * `terminate()` can return no matter how long the close takes. It is what
+ * turns "did close happen before terminate returned?" from a nanosecond race
+ * into a decided question: an implementation that completed the future first
+ * would hand the assertion a trace with no `Closed` in it.
+ *
+ * `flush()` records only when something has been written since the last
+ * flush. The real writer's `flush()` is `Output.flush()`, a no-op on an empty
+ * buffer, so a flush with nothing buffered changes no durability and is not
+ * worth a trace entry — and leaving it out keeps the trace independent of how
+ * the writer thread happens to split its drain batches.
+ */
+ class RecordingRecordWriter(recorder: Recorder)
+ extends SequentialRecordWriter[ReplayLogRecord](
+ new DataOutputStream(new ByteArrayOutputStream())
+ ) {
+ // Touched only by the writer thread (writeRecord/flush/close all run there).
+ private var buffered = false
+
+ override def writeRecord(obj: ReplayLogRecord): Unit = {
+ buffered = true
+ recorder.record(Wrote(obj))
+ }
+
+ override def flush(): Unit =
+ if (buffered) {
+ buffered = false
+ recorder.record(Flushed)
+ }
+
+ override def close(): Unit = {
+ Thread.sleep(CloseDelayMillis)
+ recorder.record(Closed)
+ }
+ }
+
+ /** See `RecordingRecordWriter.close()`. */
+ val CloseDelayMillis = 250L
+
+ /**
+ * `AsyncReplayLogWriter.terminate()` blocks on an untimed
+ * `CompletableFuture.get()` that is only completed at the very end of
+ * `run()`. amber suites are strictly serial (`Tags.limit(Tags.Test, 1)`), so
+ * a writer that never finishes would hang the whole module build rather than
+ * fail one test. Run the shutdown on a daemon thread and bound the wait, so a
+ * wedge surfaces as an ordinary failed assertion instead.
+ *
+ * The trace is captured on the shutdown thread the instant `shutdown()`
+ * returns, so assertions see exactly what the writer had done by the time the
+ * caller was released — not whatever it managed to do afterwards while the
+ * test thread was getting around to reading.
+ *
+ * @return (whether the shutdown returned inside the budget, the trace as of
+ * the moment it returned).
+ */
+ def terminatesWithin(timeoutMillis: Long, recorder: Recorder)(
+ shutdown: () => Unit
+ ): (Boolean, List[WriterEvent]) = {
+ val captured = new AtomicReference[List[WriterEvent]](Nil)
+ val t = new Thread(() => {
+ try {
+ shutdown()
+ captured.set(recorder.snapshot)
+ } catch { case _: Throwable => () }
+ })
+ t.setDaemon(true)
+ t.start()
+ t.join(timeoutMillis)
+ (!t.isAlive, captured.get())
Review Comment:
If `shutdown()` wedges, this helper returns `false` but leaves the shutdown thread running indefinitely (it’s daemon, but it can still keep interacting with shared state). It would be safer to interrupt the thread when the timeout elapses (and ensure that interruption is propagated rather than swallowed), so a wedged shutdown doesn’t leave runaway background activity in the same JVM.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogWriterFixtures.scala:
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.logreplay
+
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.MainThreadDelegateMessage
+import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage
+import org.apache.texera.amber.engine.common.storage.SequentialRecordStorage.SequentialRecordWriter
+
+import java.io.{ByteArrayOutputStream, DataOutputStream}
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.{ConcurrentLinkedQueue, CountDownLatch, TimeUnit}
+import scala.jdk.CollectionConverters.CollectionHasAsScala
+
+/**
+ * In-process fixtures shared by `AsyncReplayLogWriterSpec` and
+ * `ReplayLogManagerImplSpec`, the two specs that drive a real
+ * `AsyncReplayLogWriter` thread. Nothing here touches the filesystem, an
+ * ActorSystem, or `AmberRuntime.serde`.
+ */
+object ReplayLogWriterFixtures {
+
+ /** Ordered trace of everything the writer thread did, in the order it did it. */
+ sealed trait WriterEvent
+ final case class Wrote(record: ReplayLogRecord) extends WriterEvent
+ final case class Sent(msg: Either[MainThreadDelegateMessage, WorkflowFIFOMessage])
+ extends WriterEvent
+
+ /**
+ * A flush that actually made previously written records durable. Without this
+ * event a trace cannot tell `write; flush; send` from `write; send; flush`,
+ * and the second is a durability inversion: the message is on the network
+ * while its log record is still in the `DataOutputStream` buffer.
+ */
+ case object Flushed extends WriterEvent
+ case object Closed extends WriterEvent
+
+ /**
+ * Thread-safe recorder. The writer thread appends; the test thread reads.
+ * `ConcurrentLinkedQueue` supplies both FIFO ordering and the happens-before
+ * edge, so no extra synchronisation is needed on the assertion side.
+ */
+ class Recorder {
+ private val events = new ConcurrentLinkedQueue[WriterEvent]()
+ private val firstEvent = new CountDownLatch(1)
+
+ def record(event: WriterEvent): Unit = {
+ events.add(event)
+ firstEvent.countDown()
+ }
+
+ /** Blocks until the writer thread produces its first event, or the timeout elapses. */
+ def awaitFirstEvent(timeoutMillis: Long): Boolean =
+ firstEvent.await(timeoutMillis, TimeUnit.MILLISECONDS)
+
+ def snapshot: List[WriterEvent] = events.asScala.toList
+
+ /** The `handler` an AsyncReplayLogWriter calls for each released output. */
+ val handler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit =
+ msg => record(Sent(msg))
+ }
+
+ /**
+ * A `SequentialRecordWriter` that records rather than serialises. Overriding
+ * `writeRecord` keeps `AmberRuntime.serde` (and therefore a Pekko
+ * `ActorSystem`) out of these specs entirely; the wrapped stream is never
+ * written to and the base class's `lazy val output` is never forced.
+ *
+ * `close()` sleeps before recording. Under pristine code that costs nothing
+ * in correctness terms — `run()` closes the writer and only then completes
+ * the future `terminate()` waits on, so the `Closed` event is ordered before
+ * `terminate()` can return no matter how long the close takes. It is what
+ * turns "did close happen before terminate returned?" from a nanosecond race
+ * into a decided question: an implementation that completed the future first
+ * would hand the assertion a trace with no `Closed` in it.
+ *
+ * `flush()` records only when something has been written since the last
+ * flush. The real writer's `flush()` is `Output.flush()`, a no-op on an empty
+ * buffer, so a flush with nothing buffered changes no durability and is not
+ * worth a trace entry — and leaving it out keeps the trace independent of how
+ * the writer thread happens to split its drain batches.
+ */
+ class RecordingRecordWriter(recorder: Recorder)
+ extends SequentialRecordWriter[ReplayLogRecord](
+ new DataOutputStream(new ByteArrayOutputStream())
+ ) {
+ // Touched only by the writer thread (writeRecord/flush/close all run there).
+ private var buffered = false
+
+ override def writeRecord(obj: ReplayLogRecord): Unit = {
+ buffered = true
+ recorder.record(Wrote(obj))
+ }
+
+ override def flush(): Unit =
+ if (buffered) {
+ buffered = false
+ recorder.record(Flushed)
+ }
+
+ override def close(): Unit = {
+ Thread.sleep(CloseDelayMillis)
+ recorder.record(Closed)
+ }
Review Comment:
Using a fixed `Thread.sleep` in `close()` adds deterministic latency to every test that shuts down a writer and can noticeably slow the suite as it grows. A more efficient approach is to use synchronization primitives (e.g., a latch/future) to make the ordering testable without paying a wall-clock delay on every run, or to apply the delay only in the single test that needs it.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/AsyncReplayLogWriterSpec.scala:
##########
@@ -0,0 +1,215 @@
+/*
+ * 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.logreplay
+
+import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity}
+import org.apache.texera.amber.engine.architecture.logreplay.ReplayLogWriterFixtures.{
+ Closed,
+ Flushed,
+ Recorder,
+ RecordingRecordWriter,
+ Sent,
+ WriterEvent,
+ Wrote,
+ terminatesWithin
+}
+import org.apache.texera.amber.engine.architecture.worker.WorkflowWorker.MainThreadDelegateMessage
+import org.apache.texera.amber.engine.common.ambermessage.{DataFrame, WorkflowFIFOMessage}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class AsyncReplayLogWriterSpec extends AnyFlatSpec {
+
+ private val channel =
+ ChannelIdentity(ActorVirtualIdentity("from"), ActorVirtualIdentity("to"), isControl = false)
+
+ private def fifo(seq: Long): WorkflowFIFOMessage =
+ WorkflowFIFOMessage(channel, seq, DataFrame(Array.empty))
+
+ private def output(seq: Long): Either[MainThreadDelegateMessage, WorkflowFIFOMessage] =
+ Right(fifo(seq))
+
+ /**
+ * The other arm of the queue element's `Either`. `DPThread`,
+ * `PrepareCheckpointHandler` and `FinalizeCheckpointHandler` all push
+ * `Left(MainThreadDelegateMessage(...))` through this writer, so dropping one
+ * would break checkpointing silently rather than loudly.
+ */
+ private def delegate(): Either[MainThreadDelegateMessage, WorkflowFIFOMessage] =
+ Left(MainThreadDelegateMessage(_ => ()))
+
+ private def newWriter(recorder: Recorder): AsyncReplayLogWriter = {
+ val writer = new AsyncReplayLogWriter(recorder.handler, new RecordingRecordWriter(recorder))
+ // Daemon so that even a wedged writer can never keep a JVM from exiting.
+ writer.setDaemon(true)
+ writer
+ }
+
+ private val shutdownBudgetMillis = 30000L
+
+ /**
+ * Shuts the writer down and returns the trace as of the instant
+ * `terminate()` returned. Bounded, because `terminate()` blocks on an untimed
+ * future (see `ReplayLogWriterFixtures.terminatesWithin`).
+ */
+ private def shutdownTrace(
+ writer: AsyncReplayLogWriter,
+ recorder: Recorder
+ ): List[WriterEvent] = {
+ val (returned, trace) =
+ terminatesWithin(shutdownBudgetMillis, recorder)(() => writer.terminate())
+ assert(returned, "terminate() did not return")
+ trace
+ }
+
+ /**
+ * `logInterval` is copied from `ApplicationConfig.faultToleranceLogFlushIntervalInMs`,
+ * a memoised `val` on a shared object that must NOT be poked: amber suites run
+ * serially in one JVM, so mutating it would poison every later suite. The writer
+ * stores its own copy in a `private final long` *instance* field, and overwriting
+ * that is instance-local. Done before `start()`, so `Thread.start()`'s
+ * happens-before edge publishes the new value to the writer thread.
+ */
+ private def setLogInterval(writer: AsyncReplayLogWriter, millis: Long): Unit = {
+ val field = classOf[AsyncReplayLogWriter].getDeclaredField("logInterval")
+ field.setAccessible(true)
+ field.setLong(writer, millis)
+ assert(field.getLong(writer) == millis, "reflective logInterval override did not take effect")
+ }
+
+ /** A writer that has completed its full start/terminate lifecycle. */
+ private def startedAndTerminated(): AsyncReplayLogWriter = {
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ writer.start()
+ shutdownTrace(writer, recorder)
+ writer
+ }
+
+ "AsyncReplayLogWriter" should
+ "write and flush queued log records before releasing the queued outputs" in {
+ // The class exists to guarantee this ordering: a message must not reach the
+ // network before the log record that would let a replay reproduce it — and
+ // "written" here has to mean flushed, not merely handed to the stream. Both
+ // items are queued *before* start(), so the writer thread's first drainTo is
+ // guaranteed to pick up both in one batch, which is exactly the batch in
+ // which the ordering can be got wrong.
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ val step = ProcessingStep(channel, 0L)
+ val released = output(1L)
+
+ writer.putLogRecords(Array(step))
+ writer.putOutput(released)
+ writer.start()
+
+ assert(shutdownTrace(writer, recorder) == List(Wrote(step), Flushed, Sent(released), Closed))
+ }
+
+ it should "write every queued record, in queue order, before the outputs that follow them" in {
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ val first = ProcessingStep(channel, 0L)
+ val second = MessageContent(fifo(7L))
+ val released = output(2L)
+
+ writer.putLogRecords(Array(first, second))
+ writer.putOutput(released)
+ writer.start()
+
+ assert(
+ shutdownTrace(writer, recorder) ==
+ List(Wrote(first), Wrote(second), Flushed, Sent(released), Closed)
+ )
+ }
+
+ it should "release main-thread delegates as well as FIFO messages, in queue order" in {
+ // The queue element is an Either precisely so that main-thread delegates
+ // (checkpoint closures) ride the same ordering guarantee as network
+ // messages. A writer that released only the Right arm would drop every
+ // checkpoint closure while still writing its log records.
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ val step = ProcessingStep(channel, 0L)
+ val closure = delegate()
+ val released = output(3L)
+
+ writer.putLogRecords(Array(step))
+ writer.putOutput(closure)
+ writer.putOutput(released)
+ writer.start()
+
+ assert(
+ shutdownTrace(writer, recorder) ==
+ List(Wrote(step), Flushed, Sent(closure), Sent(released), Closed)
+ )
+ }
+
+ it should "close the underlying record writer exactly once when it shuts down" in {
+ // terminate() is the only shutdown path, and it must both stop the thread
+ // and close the record writer — a writer left open would leak the log file
+ // handle for the rest of the worker's life. Asserting on the trace captured
+ // the instant terminate() returned is what makes this an ordering claim
+ // rather than a race: an implementation that released the caller before
+ // closing is caught even though it does eventually close.
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ writer.start()
+
+ assert(shutdownTrace(writer, recorder) == List(Closed))
+ }
+
+ it should "reject further log records once it has been terminated" in {
+ val writer = startedAndTerminated()
+ intercept[AssertionError] {
+ writer.putLogRecords(Array(ProcessingStep(channel, 0L)))
+ }
+ }
+
+ it should "reject further outputs once it has been terminated" in {
+ val writer = startedAndTerminated()
+ intercept[AssertionError] {
+ writer.putOutput(output(1L))
+ }
+ }
+
+ it should "wait for the configured flush interval before draining the queue" in {
+ // With a positive faultToleranceLogFlushIntervalInMs the writer batches by
+ // sleeping at the top of every drain loop. The default is 0 (no sleep), so
+ // this arm is only reachable with a non-default interval.
+ val recorder = new Recorder
+ val writer = newWriter(recorder)
+ val flushIntervalMillis = 300L
+ setLogInterval(writer, flushIntervalMillis)
+ val step = ProcessingStep(channel, 0L)
+ writer.putLogRecords(Array(step))
+
+ val startedAt = System.nanoTime()
+ writer.start()
+ assert(recorder.awaitFirstEvent(shutdownBudgetMillis), "the queued record was never written")
+ val elapsedMillis = (System.nanoTime() - startedAt) / 1000000L
+
+ assert(shutdownTrace(writer, recorder) == List(Wrote(step), Flushed, Closed))
+ assert(
+ elapsedMillis >= 250L,
+ s"the first flush landed after only ${elapsedMillis}ms, so the ${flushIntervalMillis}ms " +
+ "flush interval was not honoured"
+ )
Review Comment:
This test hard-codes `250L` even though it’s specifically validating `flushIntervalMillis = 300L`. If `flushIntervalMillis` is changed, the assertion may no longer reflect the intended contract. Consider deriving the lower bound from `flushIntervalMillis` (optionally with an explicit tolerance) so the test stays aligned with the configured interval.
--
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): cover the replay-log writer, manager and congestion control [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #8177:
URL: https://github.com/apache/texera/pull/8177#issuecomment-5472461015
<!-- texera-reviewer-suggestion -->
### Automated Reviewer Suggestions
Based on the `git blame` history of the changed files, we recommend the following reviewers:
- **Contributors with relevant context:** `@Yicong-Huang`
You can notify them by mentioning `@Yicong-Huang` in a comment.
--
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): cover the replay-log writer, manager and congestion control [texera]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #8177:
URL: https://github.com/apache/texera/pull/8177#issuecomment-5472536690
## [Codecov](https://app.codecov.io/gh/apache/texera/pull/8177?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 93.96%. Comparing base ([`70c2114`](https://app.codecov.io/gh/apache/texera/commit/70c21145887920528d7d5540e3fb790b43e8b759?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`7768848`](https://app.codecov.io/gh/apache/texera/commit/7768848e5d5543e76a0ecdc93e6774853cd93585?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:white_check_mark: All tests successful. No failed tests found.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #8177 +/- ##
============================================
+ Coverage 93.94% 93.96% +0.01%
- Complexity 4759 4763 +4
============================================
Files 1185 1185
Lines 48069 48069
Branches 5359 5359
============================================
+ Hits 45159 45167 +8
+ Misses 1480 1476 -4
+ Partials 1430 1426 -4
```
| [Flag](https://app.codecov.io/gh/apache/texera/pull/8177/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/8177/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.00% <ø> (ø)` | | Carriedforward from [70c2114](https://app.codecov.io/gh/apache/texera/commit/70c21145887920528d7d5540e3fb790b43e8b759?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/8177/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `99.32% <ø> (ø)` | | Carriedforward from [70c2114](https://app.codecov.io/gh/apache/texera/commit/70c21145887920528d7d5540e3fb790b43e8b759?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/8177/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.03% <ø> (+0.04%)` | :arrow_up: | |
| [computing-unit-managing-service](https://app.codecov.io/gh/apache/texera/pull/8177/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.67% <ø> (ø)` | | Carriedforward from [70c2114](https://app.codecov.io/gh/apache/texera/commit/70c21145887920528d7d5540e3fb790b43e8b759?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/8177/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `86.86% <ø> (ø)` | | Carriedforward from [70c2114](https://app.codecov.io/gh/apache/texera/commit/70c21145887920528d7d5540e3fb790b43e8b759?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/8177/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `87.91% <ø> (ø)` | | Carriedforward from [70c2114](https://app.codecov.io/gh/apache/texera/commit/70c21145887920528d7d5540e3fb790b43e8b759?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/8177/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `96.49% <ø> (ø)` | | Carriedforward from [70c2114](https://app.codecov.io/gh/apache/texera/commit/70c21145887920528d7d5540e3fb790b43e8b759?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/8177/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `79.31% <ø> (ø)` | | Carriedforward from [70c2114](https://app.codecov.io/gh/apache/texera/commit/70c21145887920528d7d5540e3fb790b43e8b759?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/8177/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `98.87% <ø> (ø)` | | Carriedforward from [70c2114](https://app.codecov.io/gh/apache/texera/commit/70c21145887920528d7d5540e3fb790b43e8b759?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/8177/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `77.19% <ø> (ø)` | | Carriedforward from [70c2114](https://app.codecov.io/gh/apache/texera/commit/70c21145887920528d7d5540e3fb790b43e8b759?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/8177?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): cover the replay-log writer, manager and congestion control [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan merged PR #8177:
URL: https://github.com/apache/texera/pull/8177
--
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): cover the replay-log writer, manager and congestion control [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #8177:
URL: https://github.com/apache/texera/pull/8177#issuecomment-5472496760
<!-- texera-benchmarks-comment -->
## ⚠️ Benchmark changes need a look
🟢 2 better · 🔴 5 worse · ⚪ 8 noise (<±5%) · 0 without baseline
> Compared against main `70c2114` 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/33346622105)
| | config | throughput | MB/s | latency | max Δ latest / 7d |
|---|---|---:|---:|---:|---:|
| 🔴 | bs=10 sw=10 sl=64 | 501 | 0.306 | 19,426/24,558/24,558 us | 🟢 -34.6% / 🔴 +66.1% |
| 🔴 | bs=100 sw=10 sl=64 | 1,004 | 0.613 | 96,280/151,522/151,522 us | 🔴 +8.7% / 🔴 +50.1% |
| 🔴 | bs=1000 sw=10 sl=64 | 1,191 | 0.727 | 831,608/1,008,130/1,008,130 us | 🔴 +12.7% / 🟢 -10.0% |
<details><summary>Baseline details</summary>
Latest main `70c2114` from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---:|---:|---:|---:|---:|
| bs=10 sw=10 sl=64 | throughput | 501 tuples/sec | 503 tuples/sec | 844.57 tuples/sec | -0.4% | -40.7% |
| bs=10 sw=10 sl=64 | MB/s | 0.306 MB/s | 0.307 MB/s | 0.515 MB/s | -0.3% | -40.6% |
| bs=10 sw=10 sl=64 | p50 | 19,426 us | 16,558 us | 11,837 us | +17.3% | +64.1% |
| bs=10 sw=10 sl=64 | p95 | 24,558 us | 37,545 us | 14,788 us | -34.6% | +66.1% |
| bs=10 sw=10 sl=64 | p99 | 24,558 us | 37,545 us | 17,651 us | -34.6% | +39.1% |
| bs=100 sw=10 sl=64 | throughput | 1,004 tuples/sec | 1,043 tuples/sec | 1,084 tuples/sec | -3.7% | -7.4% |
| bs=100 sw=10 sl=64 | MB/s | 0.613 MB/s | 0.637 MB/s | 0.662 MB/s | -3.8% | -7.4% |
| bs=100 sw=10 sl=64 | p50 | 96,280 us | 92,212 us | 94,505 us | +4.4% | +1.9% |
| bs=100 sw=10 sl=64 | p95 | 151,522 us | 139,445 us | 100,941 us | +8.7% | +50.1% |
| bs=100 sw=10 sl=64 | p99 | 151,522 us | 139,445 us | 108,642 us | +8.7% | +39.5% |
| bs=1000 sw=10 sl=64 | throughput | 1,191 tuples/sec | 1,219 tuples/sec | 1,119 tuples/sec | -2.3% | +6.4% |
| bs=1000 sw=10 sl=64 | MB/s | 0.727 MB/s | 0.744 MB/s | 0.683 MB/s | -2.3% | +6.4% |
| bs=1000 sw=10 sl=64 | p50 | 831,608 us | 818,644 us | 924,404 us | +1.6% | -10.0% |
| bs=1000 sw=10 sl=64 | p95 | 1,008,130 us | 894,622 us | 967,785 us | +12.7% | +4.2% |
| bs=1000 sw=10 sl=64 | p99 | 1,008,130 us | 894,622 us | 995,630 us | +12.7% | +1.3% |
</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,399.18,200,128000,501,0.306,19425.85,24558.04,24558.04
1,100,10,64,20,1991.38,2000,1280000,1004,0.613,96280.10,151521.81,151521.81
2,1000,10,64,20,16793.75,20000,12800000,1191,0.727,831608.36,1008130.33,1008130.33
```
</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): cover the replay-log writer, manager and congestion control [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on PR #8177:
URL: https://github.com/apache/texera/pull/8177#issuecomment-5479746104
Thanks for the review — four of the five applied, one refused, plus one thing the comments didn't catch. Pushed as `be53e5c4` and `1ce599bd`.
**Applied.**
- **`ReplayLogWriterFixtures:154`** (swallowed `Throwable`) — `terminatesWithin` now stores what `shutdown()` threw in an `AtomicReference` and rethrows the original instance on the calling thread once `join()` returns. Renamed the thread local `t` → `shutdownThread` so the catch binding no longer shadows it.
- **`:155`** (wedged shutdown leaves the thread running) — after `join(timeoutMillis)` it computes `returned = !shutdownThread.isAlive`, calls `interrupt()` when the budget elapsed, and only consults the captured throwable when the thread actually returned.
- **`AsyncReplayLogWriterSpec:213`** (hard-coded `250L`) — added a named `sleepGranularityToleranceMillis = 50L` and derived the bound as `flushIntervalMillis - tolerance`. The failure message now names both the floor and the interval it came from.
- **`CongestionControlSpec:191`** (full-string `getStatusReport` assertion) — premise accepted, remedy adjusted. The clamp test now asserts the `canSend` boundary directly (`getInTransitMessages.isEmpty` and `canSend`, then `markMessageInTransit` and `!canSend`, i.e. `windowSize == 1` exactly, twice). `getStatusReport` is still built, but as the failure clue rather than the assertion.
**Refused: `ReplayLogWriterFixtures:120`, the fixed `Thread.sleep` in `close()`.** That sleep isn't incidental latency — it's the device that makes the durability ordering decidable. `AsyncReplayLogWriter.run()` closes the record writer and only *then* completes the future `terminate()` waits on, and the fixture snapshots the trace on the shutdown thread the instant `terminate()` returns. Without a delay in `close()`, "did the writer finish flushing before it signalled completion?" stops being an observable question and the ordering assertions become vacuous rather than fast. A latch would let the fixture *observe* the ordering but not *require* it.
**One thing the review didn't surface, found while re-verifying.** The dedicated `getStatusReport` format test drove all three counters to `1` — `current window size = 1 \t in transit = 1 \t waiting = 1`. A string in which every field holds the same value can't distinguish the fields from one another, so a mutant transposing two of them survived. The counters are now driven to three distinct values (`2 / 1 / 3`: one in-window ack doubles the window via slow start, one `markMessageInTransit`, three `enqueueMessage`), which is what the comment's underlying concern was really pointing at.
33/33 tests pass, lint clean, production diff empty, and the no-hang property of the shutdown helper is preserved — every test still starts before terminating, with `terminate()` in a `finally`.
--
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