Copilot commented on code in PR #7872:
URL: https://github.com/apache/texera/pull/7872#discussion_r3837767718
##########
amber/src/test/scala/org/apache/texera/web/WorkflowLifecycleManagerSpec.scala:
##########
@@ -69,24 +76,124 @@ class WorkflowLifecycleManagerSpec extends AnyFlatSpec
with BeforeAndAfterAll {
}
private def managerWithCallback(
- cleanUpTimeout: Int = 1
+ cleanUpTimeout: Int = 1,
+ id: String = "workflow-lifecycle-manager-spec"
): (WorkflowLifecycleManager, CountDownLatch) = {
val cleaned = new CountDownLatch(1)
val manager = new WorkflowLifecycleManager(
- id = "workflow-lifecycle-manager-spec",
+ id = id,
cleanUpTimeout = cleanUpTimeout,
cleanUpCallback = () => cleaned.countDown()
)
(manager, cleaned)
}
+ /** Counts the clean-ups instead of latching on the first one, so a repeat
is observable. */
+ private def managerWithCounter(
+ cleanUpTimeout: Int = 1,
+ id: String = "workflow-lifecycle-manager-spec"
+ ): (WorkflowLifecycleManager, AtomicInteger) = {
+ val cleanUps = new AtomicInteger(0)
+ val manager = new WorkflowLifecycleManager(
+ id = id,
+ cleanUpTimeout = cleanUpTimeout,
+ cleanUpCallback = () => { cleanUps.incrementAndGet(); () }
+ )
+ (manager, cleanUps)
+ }
+
private def assertCleanUpWithin(cleaned: CountDownLatch, seconds: Long):
Unit = {
assert(
cleaned.await(seconds, TimeUnit.SECONDS),
"cleanup callback was not invoked before the deadline"
)
}
+ /**
+ * Buffers the events a logger emits. logback's own ListAppender collects
into a plain
+ * ArrayList, and the clean-up body runs on the scheduler's execution
context rather than on
+ * the test thread, so the buffer has to tolerate appends from another
thread.
+ */
+ private final class CollectingAppender extends AppenderBase[ILoggingEvent] {
+ private val events = new ConcurrentLinkedQueue[ILoggingEvent]
+
+ override def append(event: ILoggingEvent): Unit = events.add(event)
+
+ def messages: Seq[String] = events.asScala.map(_.getFormattedMessage).toSeq
+ }
+
+ private val managerLoggerName = classOf[WorkflowLifecycleManager].getName
+
+ /**
+ * Runs `body` with the manager's logger raised to INFO, handing it a live
view of the
+ * messages logged for `id`.
+ *
+ * The manager reports every lifecycle decision at INFO, and amber's
logback.xml pins
+ * `org.apache` to WARN, so those calls sit behind `isInfoEnabled` and stay
dormant unless a
+ * test raises the level. Additivity is switched off so the captured lines
do not also reach
+ * the shared console and rolling-file appenders.
+ */
+ private def withInfoLogs[T](id: String)(body: (() => Seq[String]) => T): T =
{
+ val logger =
LoggerFactory.getLogger(managerLoggerName).asInstanceOf[LogbackLogger]
+ val appender = new CollectingAppender
+ val previousLevel = logger.getLevel
+ val previousAdditive = logger.isAdditive
+ appender.setContext(logger.getLoggerContext)
+ appender.setName("workflow-lifecycle-manager-spec-appender")
+ appender.start()
+ logger.addAppender(appender)
+ logger.setLevel(Level.INFO)
+ logger.setAdditive(false)
+ try {
+ body(() => appender.messages.filter(_.startsWith(s"[$id] ")))
+ } finally {
+ logger.setAdditive(previousAdditive)
+ logger.setLevel(previousLevel)
+ logger.detachAppender(appender)
+ appender.stop()
+ }
+ }
+
+ /** Waits for `fragment` to show up, then returns everything logged so far.
*/
+ private def awaitMessage(
+ messages: () => Seq[String],
+ fragment: String,
+ seconds: Long
+ ): Seq[String] = {
+ val deadline = System.currentTimeMillis() + seconds * 1000
+ while (System.currentTimeMillis() < deadline &&
!messages().exists(_.contains(fragment))) {
+ Thread.sleep(50)
+ }
+ val captured = messages()
Review Comment:
`awaitMessage` and `awaitCleanUpCount` use `System.currentTimeMillis()` to
compute deadlines. Wall-clock time can jump (e.g., NTP adjustments), which can
make these polling loops flaky (terminate early or run much longer than
intended). Using `System.nanoTime()` (monotonic) for elapsed-time deadlines
avoids that class of flake.
This issue also appears on line 175 of the same file.
##########
amber/src/test/scala/org/apache/texera/web/WorkflowLifecycleManagerSpec.scala:
##########
@@ -69,24 +76,124 @@ class WorkflowLifecycleManagerSpec extends AnyFlatSpec
with BeforeAndAfterAll {
}
private def managerWithCallback(
- cleanUpTimeout: Int = 1
+ cleanUpTimeout: Int = 1,
+ id: String = "workflow-lifecycle-manager-spec"
): (WorkflowLifecycleManager, CountDownLatch) = {
val cleaned = new CountDownLatch(1)
val manager = new WorkflowLifecycleManager(
- id = "workflow-lifecycle-manager-spec",
+ id = id,
cleanUpTimeout = cleanUpTimeout,
cleanUpCallback = () => cleaned.countDown()
)
(manager, cleaned)
}
+ /** Counts the clean-ups instead of latching on the first one, so a repeat
is observable. */
+ private def managerWithCounter(
+ cleanUpTimeout: Int = 1,
+ id: String = "workflow-lifecycle-manager-spec"
+ ): (WorkflowLifecycleManager, AtomicInteger) = {
+ val cleanUps = new AtomicInteger(0)
+ val manager = new WorkflowLifecycleManager(
+ id = id,
+ cleanUpTimeout = cleanUpTimeout,
+ cleanUpCallback = () => { cleanUps.incrementAndGet(); () }
+ )
+ (manager, cleanUps)
+ }
+
private def assertCleanUpWithin(cleaned: CountDownLatch, seconds: Long):
Unit = {
assert(
cleaned.await(seconds, TimeUnit.SECONDS),
"cleanup callback was not invoked before the deadline"
)
}
+ /**
+ * Buffers the events a logger emits. logback's own ListAppender collects
into a plain
+ * ArrayList, and the clean-up body runs on the scheduler's execution
context rather than on
+ * the test thread, so the buffer has to tolerate appends from another
thread.
+ */
+ private final class CollectingAppender extends AppenderBase[ILoggingEvent] {
+ private val events = new ConcurrentLinkedQueue[ILoggingEvent]
+
+ override def append(event: ILoggingEvent): Unit = events.add(event)
+
+ def messages: Seq[String] = events.asScala.map(_.getFormattedMessage).toSeq
+ }
+
+ private val managerLoggerName = classOf[WorkflowLifecycleManager].getName
+
+ /**
+ * Runs `body` with the manager's logger raised to INFO, handing it a live
view of the
+ * messages logged for `id`.
+ *
+ * The manager reports every lifecycle decision at INFO, and amber's
logback.xml pins
+ * `org.apache` to WARN, so those calls sit behind `isInfoEnabled` and stay
dormant unless a
+ * test raises the level. Additivity is switched off so the captured lines
do not also reach
+ * the shared console and rolling-file appenders.
+ */
+ private def withInfoLogs[T](id: String)(body: (() => Seq[String]) => T): T =
{
+ val logger =
LoggerFactory.getLogger(managerLoggerName).asInstanceOf[LogbackLogger]
+ val appender = new CollectingAppender
+ val previousLevel = logger.getLevel
+ val previousAdditive = logger.isAdditive
+ appender.setContext(logger.getLoggerContext)
+ appender.setName("workflow-lifecycle-manager-spec-appender")
+ appender.start()
+ logger.addAppender(appender)
+ logger.setLevel(Level.INFO)
+ logger.setAdditive(false)
+ try {
+ body(() => appender.messages.filter(_.startsWith(s"[$id] ")))
+ } finally {
+ logger.setAdditive(previousAdditive)
+ logger.setLevel(previousLevel)
+ logger.detachAppender(appender)
+ appender.stop()
+ }
+ }
Review Comment:
`withInfoLogs` mutates the global Logback logger (level + additivity) for
`WorkflowLifecycleManager`. Since the amber module can run suites concurrently
in the same JVM, concurrent tests that also instantiate
`WorkflowLifecycleManager` can have their logs suppressed or have logger state
restored incorrectly if two captures overlap. Wrapping the mutation/restore in
a `logger.synchronized` block (and detaching the appender before restoring
state) makes the logger state changes atomic and avoids interleaving restores.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
