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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
