This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7674-256c2d10471a5182358fd6783c4680adbfc5e2c0 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 1c4662b7cff853bdb06414f973956a09dd156448 Author: Xinyuan Lin <[email protected]> AuthorDate: Mon Aug 17 06:00:43 2026 +0000 refactor(amber): collapse the single-subclass ReplayLogger hierarchy into one class (#7674) ### What changes were proposed in this PR? Collapses the two-file `ReplayLogger` hierarchy into one concrete class. #7452 removed `EmptyReplayLogger`, which left the abstract base with a single subclass and no declared-type site anywhere: | `ReplayLogger` | Subclasses | Declared-type sites | | --- | ---: | --- | | before #7452 | 2 | 1 — `EmptyReplayLoggerSpec.scala:125`, itself deleted by #7452 | | on `main` today | 1 | 0 | ``` ReplayLogManager.scala:109 private val replayLogger = new ReplayLoggerImpl() -> infers the concrete class ReplayLoggerImpl.scala:31 class ReplayLoggerImpl extends ReplayLogger -> the only subclass ReplayLogger.scala:28 abstract class ReplayLogger -> nothing else refers to it ``` So the base declared three abstract methods that exactly one class implemented and that no call site dispatched through. Dropping it removes `extends ReplayLogger`, one `override` keyword that no longer overrides anything, and one of the two files. The three method bodies move across untouched — no behaviour change. **Naming.** The surviving class takes the plain `ReplayLogger` name. With the base gone there is no abstraction left for an `...Impl` suffix to distinguish it from, and the suffix would advertise an interface that no longer exists. `ReplayLogManagerImpl` is unaffected — it does sit beside a real `ReplayLogManager` trait in the same package. ``` before after ReplayLogger.scala abstract class ReplayLogger ReplayLogger.scala class ReplayLogger ReplayLoggerImpl.scala class ReplayLoggerImpl (deleted) extends ReplayLogger ``` The rename touches one production call site (`ReplayLogManager.scala:109`) and the `new ReplayLoggerImpl()` constructions plus test names in `LogreplayPrimitivesSpec`. It also fixes a stale `@param channel` in the Scaladoc, whose parameter has been called `channelId` all along. ### Any related issues, documentation, discussions? Closes #7673 ### How was this PR tested? Existing tests only — this is a structural change with no behaviour change, and `LogreplayPrimitivesSpec` already covers the class by constructing it directly, so it pins all three methods across the refactor. Its only edits are the type name. Locally, from the repo root with Java 17: - `sbt "WorkflowExecutionService/Test/compile"` — success. - `sbt "WorkflowExecutionService/testOnly *LogreplayPrimitivesSpec *EmptyReplayLogManagerImplSpec *ReplayLogGeneratorSpec"` — all green. - `sbt scalafmtCheckAll "scalafixAll --check"` — clean. Verification, re-runnable by a reviewer: ``` git grep -rn ReplayLoggerImpl # empty — no Impl name survives git grep -nw ReplayLogger -- '*.scala' | grep -v Spec # 2 hits: the class, and its one call site ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) --- .../architecture/logreplay/ReplayLogManager.scala | 2 +- .../architecture/logreplay/ReplayLogger.scala | 58 +++++++++++++-- .../architecture/logreplay/ReplayLoggerImpl.scala | 86 ---------------------- .../logreplay/LogreplayPrimitivesSpec.scala | 24 +++--- 4 files changed, 65 insertions(+), 105 deletions(-) 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 a894ab3c8e..b00649cf45 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 9fbd8bf1a0..216b2a0ca6 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 42f5b9e206..0000000000 --- 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 642dcb8d9a..b4d8efaf17 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.
