From d844b3bdc653402e2c8729d4061a887984b815ae Mon Sep 17 00:00:00 2001 From: Xinyuan Lin Date: Fri, 14 Aug 2026 19:01:10 -0700 Subject: [PATCH] refactor(amber): collapse the single-subclass ReplayLogger hierarchy into one class After EmptyReplayLogger was removed, the abstract base ReplayLogger had one subclass and was never named as a type: ReplayLogManagerImpl builds it as `new ReplayLoggerImpl()`, so the inferred type is the concrete class. Merges the two into a single `class ReplayLogger` in ReplayLogger.scala, dropping `extends ReplayLogger` and the now-meaningless `override`. The Impl suffix goes with the base -- it advertised an interface that no longer exists. The three method bodies are untouched; no behaviour change. Also fixes a stale `@param channel` in the Scaladoc; the parameter has been called channelId all along. --- .../logreplay/ReplayLogManager.scala | 2 +- .../architecture/logreplay/ReplayLogger.scala | 58 +++++++++++-- .../logreplay/ReplayLoggerImpl.scala | 86 ------------------- .../logreplay/LogreplayPrimitivesSpec.scala | 24 +++--- 4 files changed, 65 insertions(+), 105 deletions(-) delete mode 100644 amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLoggerImpl.scala diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogManager.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogManager.scala index a894ab3c8ef..b00649cf45c 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogManager.scala +++ b/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 = _ diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogger.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogger.scala index 9fbd8bf1a0b..216b2a0ca6b 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLogger.scala +++ b/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)) + } } diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLoggerImpl.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLoggerImpl.scala deleted file mode 100644 index 42f5b9e2069..00000000000 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/logreplay/ReplayLoggerImpl.scala +++ /dev/null @@ -1,86 +0,0 @@ -/* - * 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.{ - ChannelIdentity, - EmbeddedControlMessageIdentity -} -import org.apache.texera.amber.engine.architecture.common.ProcessingStepCursor.INIT_STEP -import org.apache.texera.amber.engine.common.ambermessage.WorkflowFIFOMessage - -import scala.collection.mutable - -class ReplayLoggerImpl extends 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 channel The channel ID associated with the processing step. - * @param message An optional message associated with the processing step. - */ - override def logCurrentStepWithMessage( - step: Long, - channelId: ChannelIdentity, - 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)) - } - } - - /** - * 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)) - } -} diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/LogreplayPrimitivesSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/LogreplayPrimitivesSpec.scala index 642dcb8d9ac..b4d8efaf176 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/logreplay/LogreplayPrimitivesSpec.scala +++ b/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.