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