This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 1c4662b7cf refactor(amber): collapse the single-subclass ReplayLogger
hierarchy into one class (#7674)
1c4662b7cf is described below
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.
