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-7028-83849efea189afc303ba074b75cd8c13731fdcff in repository https://gitbox.apache.org/repos/asf/texera.git
commit abdf440324df32e15f96f90aa1507ad19678c8d2 Author: Xinyuan Lin <[email protected]> AuthorDate: Wed Jul 29 17:11:24 2026 -0700 chore(amber): remove unused OrderingEnforcer (#7028) ### What changes were proposed in this PR? Deletes `OrderingEnforcer`, a generic helper in the amber engine with no production caller. Pure deletion, no behaviour change: **−203 lines**. | Removed | Lines | Why it is dead | | --- | ---: | --- | | `architecture/messaginglayer/OrderingEnforcer.scala` | 53 | last caller removed by #2208 | | `architecture/messaginglayer/OrderingEnforcerSpec.scala` | 150 | covers only the above | `OrderingEnforcer` was the extracted, generic form of the FIFO / exactly-once reorder buffer. #2208 ("Refactoring of amber engine", 2023-11-15) inlined that logic into `AmberFIFOChannel`, which still carries the identical header comment `/* The abstracted FIFO/exactly-once logic */` and re-declares the same members: ``` OrderingEnforcer[T] (deleted) AmberFIFOChannel (live, unchanged) ofoMap: LongMap[T] ofoMap: HashMap[Long, WorkflowFIFOMessage] current: Long current: Long isDuplicated / isAhead isDuplicated / isAhead stash / enforceFIFO stash / enforceFIFO ``` The generic copy has been stranded for two and a half years. > Reviewer note: there is a separate, **live** `OrderEnforcer` trait (no `-ing`) in `architecture/logreplay/`, used by `InputGateway`/`NetworkInputGateway`. It is unrelated and untouched — only the `messaginglayer` `OrderingEnforcer` is removed here. It acquired unit tests during the 2026 coverage work (#4721), which is why it currently looks live despite having been unreachable since 2023. **Scope change after review:** this PR originally also removed `Utils.retry`. Per @Yicong-Huang's review it has been restored and is untouched here — `Utils.scala` and `UtilsSpec.scala` are byte-identical to `main`. ### Any related issues, documentation, discussions? Closes #7027 ### How was this PR tested? Existing tests only — this PR adds none, since it removes code and the spec that covered it. Locally, from the repo root with Java 17: - `sbt "scalafixAll --check"` — clean. - `sbt scalafmtCheckAll` — clean. - `sbt "WorkflowExecutionService/testOnly *UtilsSpec"` — 74 tests, all pass (`Utils.retry` and its three tests are back and green). Verification that nothing references the removed class, re-runnable by a reviewer: ``` git grep -nw OrderingEnforcer # only the two deleted files ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) --- .../messaginglayer/OrderingEnforcer.scala | 53 -------- .../messaginglayer/OrderingEnforcerSpec.scala | 150 --------------------- 2 files changed, 203 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OrderingEnforcer.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OrderingEnforcer.scala deleted file mode 100644 index d840738a45..0000000000 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OrderingEnforcer.scala +++ /dev/null @@ -1,53 +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.messaginglayer - -import scala.collection.mutable - -/* The abstracted FIFO/exactly-once logic */ -class OrderingEnforcer[T] { - - val ofoMap = new mutable.LongMap[T] - var current = 0L - - def setCurrent(value: Long): Unit = { - current = value - } - - def isDuplicated(sequenceNumber: Long): Boolean = - sequenceNumber < current || ofoMap.contains(sequenceNumber) - - def isAhead(sequenceNumber: Long): Boolean = sequenceNumber > current - - def stash(sequenceNumber: Long, data: T): Unit = { - ofoMap(sequenceNumber) = data - } - - def enforceFIFO(data: T): List[T] = { - val res = mutable.ArrayBuffer[T](data) - current += 1 - while (ofoMap.contains(current)) { - res.append(ofoMap(current)) - ofoMap.remove(current) - current += 1 - } - res.toList - } -} diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OrderingEnforcerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OrderingEnforcerSpec.scala deleted file mode 100644 index be0e34eacf..0000000000 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OrderingEnforcerSpec.scala +++ /dev/null @@ -1,150 +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.messaginglayer - -import org.scalatest.flatspec.AnyFlatSpec -import org.scalatest.matchers.should.Matchers - -class OrderingEnforcerSpec extends AnyFlatSpec with Matchers { - - // ----- initial state ----- - - "OrderingEnforcer" should "start with current=0 and an empty stash" in { - val enforcer = new OrderingEnforcer[String] - enforcer.current shouldBe 0L - enforcer.ofoMap shouldBe empty - } - - // ----- setCurrent ----- - - "setCurrent" should "advance the current cursor and shift the duplicate threshold" in { - val enforcer = new OrderingEnforcer[String] - enforcer.setCurrent(10L) - enforcer.current shouldBe 10L - enforcer.isDuplicated(9L) shouldBe true - enforcer.isDuplicated(10L) shouldBe false - } - - // ----- isDuplicated ----- - - "isDuplicated" should "treat sequence numbers below current as duplicates" in { - val enforcer = new OrderingEnforcer[String] - enforcer.setCurrent(5L) - enforcer.isDuplicated(0L) shouldBe true - enforcer.isDuplicated(4L) shouldBe true - } - - it should "treat sequence numbers >= current that are not stashed as not duplicated" in { - val enforcer = new OrderingEnforcer[String] - enforcer.setCurrent(5L) - enforcer.isDuplicated(5L) shouldBe false - enforcer.isDuplicated(7L) shouldBe false - } - - it should "report stashed future sequence numbers as duplicated" in { - val enforcer = new OrderingEnforcer[String] - enforcer.stash(7L, "seven") - enforcer.isDuplicated(7L) shouldBe true - } - - // ----- isAhead ----- - - "isAhead" should "be true only for sequence numbers strictly greater than current" in { - val enforcer = new OrderingEnforcer[String] - enforcer.setCurrent(5L) - enforcer.isAhead(6L) shouldBe true - enforcer.isAhead(5L) shouldBe false - enforcer.isAhead(4L) shouldBe false - } - - // ----- stash ----- - - "stash" should "store data under its sequence number for later draining" in { - val enforcer = new OrderingEnforcer[String] - enforcer.stash(2L, "two") - enforcer.ofoMap(2L) shouldBe "two" - } - - it should "overwrite an existing stash entry at the same sequence number" in { - // Pin: there is no guard against re-stashing the same sequence number. - // Callers rely on isDuplicated to skip the second stash, but a direct - // re-stash still overwrites silently. - val enforcer = new OrderingEnforcer[String] - enforcer.stash(2L, "first") - enforcer.stash(2L, "second") - enforcer.ofoMap(2L) shouldBe "second" - } - - // ----- enforceFIFO ----- - - "enforceFIFO" should "advance current by one and emit just the input when no stash is queued" in { - val enforcer = new OrderingEnforcer[String] - enforcer.enforceFIFO("zero") shouldBe List("zero") - enforcer.current shouldBe 1L - } - - it should "drain a single contiguous stashed entry after the input" in { - val enforcer = new OrderingEnforcer[String] - enforcer.stash(1L, "one") - enforcer.enforceFIFO("zero") shouldBe List("zero", "one") - enforcer.current shouldBe 2L - enforcer.ofoMap should not contain key(1L) - } - - it should "drain a contiguous run from the stash and stop at the first gap" in { - val enforcer = new OrderingEnforcer[String] - enforcer.stash(1L, "one") - enforcer.stash(2L, "two") - enforcer.stash(4L, "four") // gap at 3 - val emitted = enforcer.enforceFIFO("zero") - emitted shouldBe List("zero", "one", "two") - enforcer.current shouldBe 3L - enforcer.ofoMap.keys.toList shouldBe List(4L) - } - - it should "leave the stash untouched when none of the queued entries are contiguous" in { - val enforcer = new OrderingEnforcer[String] - enforcer.stash(5L, "five") - enforcer.stash(7L, "seven") - val emitted = enforcer.enforceFIFO("zero") - emitted shouldBe List("zero") - enforcer.current shouldBe 1L - enforcer.ofoMap.keys.toSet shouldBe Set(5L, 7L) - } - - it should "respect a non-zero starting current when draining" in { - // Setting the cursor manually mimics replay/recovery: the enforcer skips - // past prior messages and only drains entries with sequence numbers - // strictly greater than the current value at call time. - val enforcer = new OrderingEnforcer[String] - enforcer.setCurrent(10L) - enforcer.stash(11L, "eleven") - enforcer.stash(12L, "twelve") - val emitted = enforcer.enforceFIFO("ten") - emitted shouldBe List("ten", "eleven", "twelve") - enforcer.current shouldBe 13L - } - - it should "support int payloads via the type parameter" in { - val enforcer = new OrderingEnforcer[Int] - enforcer.stash(1L, 100) - enforcer.enforceFIFO(0) shouldBe List(0, 100) - } -}
