refactor(amber): collapse the single-subclass ReplayLogger hierarchy into one class

amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogManager.scala

@@ -106,7 +106,7 @@ class EmptyReplayLogManagerImpl(
 class ReplayLogManagerImpl(handler: Either[MainThreadDelegateMessage, WorkflowFIFOMessage] => Unit)
     extends ReplayLogManager {
 
-  private val replayLogger = new ReplayLoggerImpl()
+  private val replayLogger = new ReplayLogger()
 
   private var writer: AsyncReplayLogWriter = _
 

amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogger.scala

@@ -23,18 +23,64 @@ import org.apache.texera.amber.core.virtualidentity.{
   ChannelIdentity,
   EmbeddedControlMessageIdentity
 }
+import org.apache.texera.amber.engine.architecture.common.ProcessingStepCursor.INIT_STEP
 import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage
 
-abstract class ReplayLogger {
+import scala.collection.mutable
 
+class ReplayLogger {
+
+  private val tempLogs = mutable.ArrayBuffer[ReplayLogRecord]()
+
+  private var currentChannelId: ChannelIdentity = _
+
+  private var lastStep = INIT_STEP
+
+  /**
+    * Records the current processing step along with an associated message.
+    * This method also monitors the channel information. If the new channel matches the last recorded channel
+    * and there is no associated message for this step, the logging operation is bypassed.
+    * Otherwise, it appends a ProcessingStep log record with the message content, provided the message exists.
+    *
+    * @param step      The current processing step.
+    * @param channelId The channel ID associated with the processing step.
+    * @param message   An optional message associated with the processing step.
+    */
   def logCurrentStepWithMessage(
       step: Long,
       channelId: ChannelIdentity,
-      msg: Option[WorkflowFIFOMessage]
-  ): Unit
-
-  def markAsReplayDestination(id: EmbeddedControlMessageIdentity): Unit
+      message: Option[WorkflowFIFOMessage]
+  ): Unit = {
+    if (currentChannelId == channelId && message.isEmpty) {
+      return
+    }
+    currentChannelId = channelId
+    lastStep = step
+    tempLogs.append(ProcessingStep(channelId, step))
+    if (message.isDefined) {
+      tempLogs.append(MessageContent(message.get))
+    }
+  }
 
-  def drainCurrentLogRecords(step: Long): Array[ReplayLogRecord]
+  /**
+    * Called when the data processor attempts to output a message.
+    * This method retrieves all accumulated log records and passes them to the writer thread for persistence.
+    * It ensures the processing up to the current processing step is captured in the log records.
+    *
+    * @param step The current processing step.
+    * @return An array of ReplayLogRecord containing all the log records up to the current step.
+    */
+  def drainCurrentLogRecords(step: Long): Array[ReplayLogRecord] = {
+    if (lastStep != step) {
+      lastStep = step
+      tempLogs.append(ProcessingStep(currentChannelId, step))
+    }
+    val result = tempLogs.toArray
+    tempLogs.clear()
+    result
+  }
 
+  def markAsReplayDestination(id: EmbeddedControlMessageIdentity): Unit = {
+    tempLogs.append(ReplayDestination(id))
+  }
 }

amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/LogreplayPrimitivesSpec.scala

@@ -89,18 +89,18 @@ class LogreplayPrimitivesSpec extends AnyFlatSpec with BeforeAndAfterAll {
     WorkflowFIFOMessage(cidA, seq, FixedSizePayload())
 
   // ---------------------------------------------------------------------------
-  // ReplayLoggerImpl
+  // ReplayLogger
   // ---------------------------------------------------------------------------
 
-  "ReplayLoggerImpl.logCurrentStepWithMessage" should "append a ProcessingStep when the channel changes" in {
-    val l = new ReplayLoggerImpl()
+  "ReplayLogger.logCurrentStepWithMessage" should "append a ProcessingStep when the channel changes" in {
+    val l = new ReplayLogger()
     l.logCurrentStepWithMessage(0L, cidA, None)
     val drained = l.drainCurrentLogRecords(0L)
     assert(drained.toList == List(ProcessingStep(cidA, 0L)))
   }
 
   it should "skip a same-channel call with no message" in {
-    val l = new ReplayLoggerImpl()
+    val l = new ReplayLogger()
     l.logCurrentStepWithMessage(0L, cidA, None)
     l.drainCurrentLogRecords(0L) // reset
     l.logCurrentStepWithMessage(1L, cidA, None) // same channel, no message
@@ -110,7 +110,7 @@ class LogreplayPrimitivesSpec extends AnyFlatSpec with BeforeAndAfterAll {
   }
 
   it should "append both a ProcessingStep and a MessageContent when a message is provided" in {
-    val l = new ReplayLoggerImpl()
+    val l = new ReplayLogger()
     val m = msg(7L)
     l.logCurrentStepWithMessage(2L, cidA, Some(m))
     val drained = l.drainCurrentLogRecords(2L)
@@ -122,7 +122,7 @@ class LogreplayPrimitivesSpec extends AnyFlatSpec with BeforeAndAfterAll {
     // && message.isEmpty` — both conditions, not just the channel match. After a
     // first call sets the current channel, a *subsequent* same-channel call with
     // a non-empty message must still emit ProcessingStep + MessageContent.
-    val l = new ReplayLoggerImpl()
+    val l = new ReplayLogger()
     l.logCurrentStepWithMessage(0L, cidA, None) // sets currentChannelId = cidA
     l.drainCurrentLogRecords(0L) // reset
     val m = msg(11L)
@@ -132,21 +132,21 @@ class LogreplayPrimitivesSpec extends AnyFlatSpec with BeforeAndAfterAll {
   }
 
   it should "append a ProcessingStep on a channel switch even if no message is provided" in {
-    val l = new ReplayLoggerImpl()
+    val l = new ReplayLogger()
     l.logCurrentStepWithMessage(0L, cidA, None)
     l.logCurrentStepWithMessage(1L, cidB, None) // channel change → must record
     val drained = l.drainCurrentLogRecords(1L)
     assert(drained.toList == List(ProcessingStep(cidA, 0L), ProcessingStep(cidB, 1L)))
   }
 
-  "ReplayLoggerImpl.markAsReplayDestination" should
+  "ReplayLogger.markAsReplayDestination" should
     "preserve exact ordering: in-flight ProcessingStep, then ReplayDestination, then synthetic trailing step" in {
     // ReplayLogGenerator depends on the relative position of ReplayDestination
     // within the record stream — replay stops at it. So a `contains` check
     // would silently accept a regression that duplicated ReplayDestination or
     // moved it after the synthetic trailing ProcessingStep emitted by drain.
     // Pin the full sequence instead.
-    val l = new ReplayLoggerImpl()
+    val l = new ReplayLogger()
     val ecm = EmbeddedControlMessageIdentity("checkpoint-1")
     l.logCurrentStepWithMessage(0L, cidA, None) // sets currentChannelId, appends ProcessingStep
     l.markAsReplayDestination(ecm)
@@ -164,8 +164,8 @@ class LogreplayPrimitivesSpec extends AnyFlatSpec with BeforeAndAfterAll {
     )
   }
 
-  "ReplayLoggerImpl.drainCurrentLogRecords" should "clear the buffer between drains" in {
-    val l = new ReplayLoggerImpl()
+  "ReplayLogger.drainCurrentLogRecords" should "clear the buffer between drains" in {
+    val l = new ReplayLogger()
     l.logCurrentStepWithMessage(0L, cidA, None)
     val first = l.drainCurrentLogRecords(0L)
     val second = l.drainCurrentLogRecords(0L)
@@ -174,7 +174,7 @@ class LogreplayPrimitivesSpec extends AnyFlatSpec with BeforeAndAfterAll {
   }
 
   it should "append a synthetic ProcessingStep when the requested step differs from lastStep" in {
-    val l = new ReplayLoggerImpl()
+    val l = new ReplayLogger()
     l.logCurrentStepWithMessage(0L, cidA, None)
     val drained = l.drainCurrentLogRecords(5L)
     // Two records: the original ProcessingStep at step 0 and the synthetic one at step 5.