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 57d4dba20a chore(amber): reduce engine log verbosity and pin the CI
log-level backstop (#6797)
57d4dba20a is described below
commit 57d4dba20ae395ccd76795308cda8b190c3c1df9
Author: Xinyuan Lin <[email protected]>
AuthorDate: Sat Jul 25 02:34:16 2026 -0700
chore(amber): reduce engine log verbosity and pin the CI log-level backstop
(#6797)
### What changes were proposed in this PR?
**Root cause.** `AmberLogging` names each logger `"<actorId>]
[<ClassName>"` so the logback pattern `[%logger]` renders both the actor
id and the class. The side effect is that these logger names start with
the **actor id** (`WF6-…`, `COORDINATOR`), not `org.apache…`, so the
`<logger name="org.apache" level="WARN"/>` rule in
`amber/src/main/resources/logback.xml` never matches the engine classes.
They all fall through to the root logger and emit at `INFO`. In
loop/integration tests, workers are recreated every iteration, so
per-ECM / per-worker / per-region lines fire dozens of times and flood
the CI console.
This PR demotes those statements to `DEBUG` so they no longer print at
the default `INFO` level (CI, local, and prod), and demotes three
`WARN`s that fire on **normal** operation:
| Message | Was | Source |
|---|---|---|
| `receive` / `process` / `send ECM` (2–3× per control message, full
`ChannelIdentity` dump) | INFO | `DataProcessor.scala`, `main_loop.py` |
| `register <id> -> Actor[…]` | INFO |
`PekkoActorRefMappingService.scala` |
| `…is not reachable anymore, it might have crashed` — fires on graceful
teardown | WARN | `PekkoActorRefMappingService.scala` |
| `unknown identifier` — registration race | WARN |
`PekkoActorRefMappingService.scala` |
| `worker replay log writing conf`, `DP thread started`/`exits`,
`Starting the worker.` | INFO | `WorkflowActor`, `DPThread`,
`StartHandler` |
| `Region N successfully terminated.` | INFO |
`RegionExecutionManager.scala` |
| `…completed, # of input ports…` | INFO | `DataProcessor.scala` |
| `client actor cannot handle …` — `//skip` fallthrough | WARN |
`ClientActor.scala` |
Genuine fault paths are **untouched**: `DP Thread exists unexpectedly`
(ERROR), `Failed to fetch actorRef` (WARN), and `Error when terminating
region` (WARN, the failure branch — distinct from the demoted
`successfully terminated` success branch).
As a CI backstop, the amber unit and integration test steps in
`build.yml` pin the log level so any remaining `INFO` chatter is
suppressed there even before the source demotions take effect. The two
logging stacks spell the level differently, which is a trap worth
calling out: logback takes `TEXERA_SERVICE_LOG_LEVEL=WARN`, but the
spawned Python UDF worker uses **loguru**, whose only warning level is
`WARNING`. Feeding loguru `WARN` raises `ValueError: Level 'WARN' does
not exist` at worker startup — before the worker hands its Flight port
back to the JVM — and the JVM then blocks on the proxy handshake with no
internal timeout, hanging `amber-integration` until GitHub's 6 h runner
cap. So the Python var is pinned to
`UDF_PYTHON_LOG_STREAMHANDLER_LEVEL=WARNING`, and, symmetrically,
logback keeps `WARN` (it falls back to `DEBUG` on an unknown level like
`WARNING`, which would flood logs instead). `amber-integration`
additionally gets `timeout-minutes: 40` so any future worker-startup
deadlock fails fast instead of burning a full runner.
Net effect: the bulk of the per-iteration output from loop/integration
specs disappears from the default-level log, while `DEBUG` keeps it
available on demand.
### Any related issues, documentation, discussions?
Closes #6796
### How was this PR tested?
The source demotions are a log-level-only change — no behavior changes —
so they rely on the existing test suites run in CI. Verified
additionally by inspection and by the CI backstop's own behavior:
- **No test asserts on a demoted message.** Grepped all amber Scala and
Python test sources for each demoted string; none is asserted.
(`ReplayLogGeneratorSpec` matches `"cannot handle"`, but against an
unrelated `RuntimeException` from `ReplayLogGenerator`, not the
`ClientActor` log.)
- **The CI `WARN`/`WARNING` env does not break config specs.**
`UdfConfigSpec`'s and `PekkoConfigSpec`'s default-value assertions are
guarded (`ifUnset` / `!sys.env.contains(...)`), and
`pekko.stdout-loglevel` is hardcoded `INFO` in `cluster.conf` (only
`pekko.loglevel` is env-driven), so its unguarded assertion still
passes.
- **loguru level names verified.** On the pinned loguru 0.7.0,
`logger.add(sys.stderr, level="WARN")` raises `ValueError: Level 'WARN'
does not exist` while `level="WARNING"` succeeds (loguru's levels are
`TRACE/DEBUG/INFO/SUCCESS/WARNING/ERROR/CRITICAL`); logback accepts
`WARN`. With `WARNING`, `amber-integration` runs green again — the
ubuntu leg finishes the integration step in ~8.5 min, versus the earlier
`WARN` runs that sat in that step ~3 h until cancelled.
- **Lint clean.** No imports added (so `scalafixAll --check` is
unaffected); the longest changed Scala line is 96 columns (< the 100
`maxColumn`), so `scalafmtCheckAll` stays green.
A follow-up sweep for other hot-path `INFO`/`WARN` logs (e.g.
`AsyncRPCClient` null control-reply, the range-shuffle partitioner,
`Tuple` cast warnings) is captured in #6796 rather than expanded here.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
---
.github/workflows/build.yml | 67 ++++++++++++++++++++++
amber/src/main/python/core/runnables/main_loop.py | 22 ++++---
.../common/PekkoActorRefMappingService.scala | 6 +-
.../engine/architecture/common/WorkflowActor.scala | 2 +-
.../scheduling/RegionExecutionManager.scala | 2 +-
.../engine/architecture/worker/DPThread.scala | 4 +-
.../engine/architecture/worker/DataProcessor.scala | 8 +--
.../worker/promisehandlers/StartHandler.scala | 2 +-
.../amber/engine/common/client/ClientActor.scala | 2 +-
common/config/src/main/resources/cluster.conf | 3 +
.../apache/texera/common/config/PekkoConfig.scala | 28 ++++++++-
.../texera/common/config/PekkoConfigSpec.scala | 40 +++++++++++++
12 files changed, 164 insertions(+), 22 deletions(-)
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 1355a4dbbc..b2feaad21d 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -291,6 +291,19 @@ jobs:
AMBER_TEST_FILTER: skip-integration
# unit job uses its provisioned postgres catalog; default (rest)
needs a Lakekeeper not run here
STORAGE_ICEBERG_CATALOG_TYPE: postgres
+ # Backstop for CI log volume: the chatty per-worker/per-message
engine
+ # logs are DEBUG in source; pin the JVM root and the spawned Python
UDF
+ # workers so nothing at INFO leaks into the CI console. Re-running
the
+ # job with "Enable debug logging" (which sets runner.debug to '1')
+ # flips both to DEBUG, bringing the full engine trace back on demand.
+ # Mind the spelling: logback's level is WARN, but loguru (the Python
+ # worker) only knows WARNING and raises ValueError on "WARN" — which
+ # crashes the worker at startup before it hands its port back,
+ # hanging the whole job. pekko.loglevel shares loguru's vocabulary;
+ # PekkoConfig.normalizePekkoLogLevel translates the logback spelling
+ # so the JVM knob can stay WARN here.
+ TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' ||
'WARN' }}
+ UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: ${{ runner.debug == '1' &&
'DEBUG' || 'WARNING' }}
run: |
sbt "DAO/jacoco" \
"Auth/jacoco" \
@@ -330,6 +343,12 @@ jobs:
# cutting lints (scalafmt / scalafix) and the amber dist + binary
# license check stay in `amber`; this job is tests-only.
if: ${{ inputs.run_amber_integration }}
+ # A Python UDF worker that fails to start (e.g. a bad log level) leaves the
+ # JVM blocked on the proxy handshake with no internal timeout, so the whole
+ # job would otherwise hang until GitHub's 6h cap. Fail fast instead: recent
+ # green runs take 9-17 minutes, so 25 leaves headroom for a cold cache or a
+ # slow macOS runner without masking a real hang.
+ timeout-minutes: 25
strategy:
# macOS provisions postgres / minio / lakekeeper natively because
# GitHub-hosted macOS runners have no Docker (and `services:`
@@ -609,6 +628,19 @@ jobs:
# by amber's unit-test coverage.
env:
AMBER_TEST_FILTER: integration-only
+ # Backstop for CI log volume: the chatty per-worker/per-message
engine
+ # logs are DEBUG in source; pin the JVM root and the spawned Python
UDF
+ # workers so nothing at INFO leaks into the CI console. Re-running
the
+ # job with "Enable debug logging" (which sets runner.debug to '1')
+ # flips both to DEBUG, bringing the full engine trace back on demand.
+ # Mind the spelling: logback's level is WARN, but loguru (the Python
+ # worker) only knows WARNING and raises ValueError on "WARN" — which
+ # crashes the worker at startup before it hands its port back,
+ # hanging the whole job. pekko.loglevel shares loguru's vocabulary;
+ # PekkoConfig.normalizePekkoLogLevel translates the logback spelling
+ # so the JVM knob can stay WARN here.
+ TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' ||
'WARN' }}
+ UDF_PYTHON_LOG_STREAMHANDLER_LEVEL: ${{ runner.debug == '1' &&
'DEBUG' || 'WARNING' }}
run: |
sbt scalafmtCheckAll \
"scalafixAll --check" \
@@ -638,6 +670,10 @@ jobs:
# root (the default) that resolves to ./amber and reads
# amber/src/main/resources/web-config.yml (port 8080).
if: matrix.os == 'ubuntu-latest'
+ env:
+ # Quiet boot logs, same wiring as the test steps above. Safe here:
+ # smoke-boot's verdict is LISTEN-based, never log-scraping (#6332).
+ TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' ||
'WARN' }}
run: .github/scripts/smoke-boot.sh
"/tmp/dists/amber-*/bin/texera-web-application" 8080
- name: Smoke-test computing-unit-master boots
# computing-unit-master boots on postgres only, like texera-web: main
@@ -651,9 +687,19 @@ jobs:
# amber/src/main/resources/computing-unit-master-config.yml (port
8085).
# Reuses the amber dist built + unzipped above for the texera-web boot.
if: matrix.os == 'ubuntu-latest'
+ env:
+ # Quiet boot logs, same wiring as the test steps above. Safe here:
+ # smoke-boot's verdict is LISTEN-based, never log-scraping (#6332).
+ TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' ||
'WARN' }}
run: .github/scripts/smoke-boot.sh
"/tmp/dists/amber-*/bin/computing-unit-master" 8085
- name: Run Python integration tests
# --junit-xml feeds the Test Analytics upload below.
+ env:
+ # Pytest runs with -s, so loguru's default stderr sink prints
+ # straight into the CI log; pin it to WARNING (loguru has no WARN)
+ # and flip to DEBUG on debug re-runs. Sinks that tests add with an
+ # explicit level are unaffected.
+ LOGURU_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARNING' }}
run: |
cd amber && pytest -m integration --junit-xml=junit-integration.xml
-sv
- name: Upload amber integration test results to Codecov
@@ -740,6 +786,12 @@ jobs:
- name: Build dist and run ${{ matrix.service }} tests with coverage
# Single sbt invocation so dist + test share compiled state. Use
# `jacoco` so the codecov upload step has a report to pick up.
+ env:
+ # CI log-volume backstop, same wiring as the amber jobs: every
+ # platform service reads this into its Dropwizard logging config
+ # (logback vocabulary, hence WARN); a re-run with "Enable debug
+ # logging" flips it to DEBUG.
+ TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' ||
'WARN' }}
run: sbt "${{ matrix.sbt_project }}/dist" "${{ matrix.sbt_project
}}/jacoco"
- name: Unzip ${{ matrix.service }} dist and check binary LICENSE +
NOTICE
# Each platform service has its own LICENSE-binary / NOTICE-binary at
@@ -925,6 +977,9 @@ jobs:
# without a runtime classpath/linkage crash (#6220).
env:
TEXERA_HOME: ${{ github.workspace }}
+ # Quiet boot logs, same wiring as the amber jobs. Safe here:
+ # smoke-boot's verdict is LISTEN-based, never log-scraping (#6332).
+ TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' ||
'WARN' }}
run: .github/scripts/smoke-boot.sh "/tmp/dists/${{ matrix.service
}}-*/bin/${{ matrix.service }}" "${{ matrix.port }}"
pyamber:
@@ -1013,6 +1068,12 @@ jobs:
# --junit-xml emits a JUnit-XML report alongside the coverage XML
# so the Test Analytics upload below can feed Codecov's failing-
# test PR comments and flaky-test detection on main.
+ env:
+ # Pytest runs with -s, so loguru's default stderr sink prints
+ # straight into the CI log; pin it to WARNING (loguru has no WARN)
+ # and flip to DEBUG on debug re-runs. Sinks that tests add with an
+ # explicit level are unaffected.
+ LOGURU_LEVEL: ${{ runner.debug == '1' && 'DEBUG' || 'WARNING' }}
run: |
cd amber && pytest -m "not integration" --cov=src/main/python
--cov-report=xml --junit-xml=junit.xml -sv
- name: Upload pyamber coverage to Codecov
@@ -1084,6 +1145,12 @@ jobs:
# --reporter=junit emits a JUnit-XML alongside coverage so the
# Test Analytics upload below can feed Codecov's failing-test PR
# comments and flaky-test detection on main.
+ env:
+ # CI log-volume backstop, same wiring as the JVM jobs. env.ts
+ # validates against ["ERROR","WARN","INFO","DEBUG"] before mapping
+ # to pino's lowercase levels, so the logback spelling WARN is the
+ # right one here too.
+ TEXERA_SERVICE_LOG_LEVEL: ${{ runner.debug == '1' && 'DEBUG' ||
'WARN' }}
run: bun test --coverage --coverage-reporter=lcov --reporter=junit
--reporter-outfile=junit.xml
- name: Upload agent-service coverage to Codecov
if: matrix.os == 'ubuntu-latest' && always()
diff --git a/amber/src/main/python/core/runnables/main_loop.py
b/amber/src/main/python/core/runnables/main_loop.py
index 29cfc2e622..d252be934e 100644
--- a/amber/src/main/python/core/runnables/main_loop.py
+++ b/amber/src/main/python/core/runnables/main_loop.py
@@ -444,8 +444,11 @@ class MainLoop(StoppableQueueBlockingRunnable):
ecm = ecm_element.payload
command = ecm.command_mapping.get(self.context.worker_id)
channel_id = self.context.current_input_channel_id
- logger.info(
- f"receive channel ECM from {channel_id}, id = {ecm.id}, cmd =
{command}"
+ logger.debug(
+ "receive channel ECM from {}, id = {}, cmd = {}",
+ channel_id,
+ ecm.id,
+ command,
)
if ecm.ecm_type != EmbeddedControlMessageType.NO_ALIGNMENT:
self.context.pause_manager.pause_input_channel(
@@ -453,8 +456,11 @@ class MainLoop(StoppableQueueBlockingRunnable):
)
if self.context.ecm_manager.is_ecm_aligned(channel_id, ecm):
- logger.info(
- f"process channel ECM from {channel_id}, id = {ecm.id}, cmd =
{command}"
+ logger.debug(
+ "process channel ECM from {}, id = {}, cmd = {}",
+ channel_id,
+ ecm.id,
+ command,
)
if command is not None:
@@ -470,9 +476,11 @@ class MainLoop(StoppableQueueBlockingRunnable):
active_channel_id
) in self.context.output_manager.get_output_channel_ids():
if active_channel_id in downstream_channels_in_scope:
- logger.info(
- f"send ECM to {active_channel_id},"
- f" id = {ecm.id}, cmd = {command}"
+ logger.debug(
+ "send ECM to {}, id = {}, cmd = {}",
+ active_channel_id,
+ ecm.id,
+ command,
)
self._send_ecm_to_channel(active_channel_id, ecm)
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala
index 2c085b9dc8..f3fa7f3b92 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/PekkoActorRefMappingService.scala
@@ -76,13 +76,13 @@ class PekkoActorRefMappingService(actorService:
PekkoActorService) extends Amber
def removeActorRef(id: ActorVirtualIdentity): Unit = {
if (actorRefMapping.contains(id)) {
val ref = actorRefMapping.remove(id).get
- logger.warn(s"actor $id is not reachable anymore, it might have crashed.
old ref = $ref")
+ logger.debug(s"actor $id is not reachable anymore. old ref = $ref")
}
}
def registerActorRef(id: ActorVirtualIdentity, ref: ActorRef): Unit = {
if (!actorRefMapping.contains(id)) {
- logger.info(s"register ${VirtualIdentityUtils.toShorterString(id)} ->
$ref")
+ logger.debug(s"register ${VirtualIdentityUtils.toShorterString(id)} ->
$ref")
actorRefMapping(id) = ref
if (messageStash.contains(id)) {
val stash = messageStash(id)
@@ -119,7 +119,7 @@ class PekkoActorRefMappingService(actorService:
PekkoActorService) extends Amber
}
} else {
// on coordinator, wait for actor ref registration.
- logger.warn(s"unknown identifier:
${VirtualIdentityUtils.toShorterString(id)}")
+ logger.debug(s"unknown identifier:
${VirtualIdentityUtils.toShorterString(id)}")
val toNotifySet = toNotifyOnRegistration.getOrElseUpdate(id,
mutable.HashSet[ActorRef]())
replyTo.foreach(toNotifySet.add)
}
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActor.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActor.scala
index 0f744dca26..ba2a539d11 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActor.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/common/WorkflowActor.scala
@@ -102,7 +102,7 @@ abstract class WorkflowActor(
val transferService: PekkoMessageTransferService =
new PekkoMessageTransferService(actorService, actorRefMappingService,
handleBackpressure)
- logger.info(s"worker replay log writing conf: $replayLogConfOpt")
+ logger.debug(s"worker replay log writing conf: $replayLogConfOpt")
val logStorage: SequentialRecordStorage[ReplayLogRecord] =
SequentialRecordStorage.getStorage(replayLogConfOpt.map(_.writeTo))
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
index fd239b59d5..586d55ca25 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
@@ -210,7 +210,7 @@ class RegionExecutionManager(
// 3. Log whether the kills were successful
gracefulStopRequests.transform {
case Return(_) =>
- logger.info(s"Region ${region.id.id} successfully terminated.")
+ logger.debug(s"Region ${region.id.id} successfully terminated.")
regionExecution.getAllOperatorExecutions.foreach {
case (_, opExec) =>
opExec.getWorkerIds.foreach { workerId =>
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala
index f845836843..829eaf1a77 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala
@@ -99,7 +99,7 @@ class DPThread(
// so create() can append a suffix without the execution id. Once
per thread,
// assuming a thread serves one execution.
LargeBinaryManager.setCurrentBaseUri(largeBinaryBaseUri)
- logger.info("DP thread started")
+ logger.debug("DP thread started")
startFuture.complete(())
dp.statisticsManager.initializeWorkerStartTime(System.nanoTime())
try {
@@ -107,7 +107,7 @@ class DPThread(
} catch safely {
case _: InterruptedException =>
// dp thread will stop here
- logger.info("DP Thread exits")
+ logger.debug("DP Thread exits")
case err: Throwable =>
logger.error("DP Thread exists unexpectedly", err)
dp.outputHandler(Left(MainThreadDelegateMessage((worker) => {
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala
index a86d16c926..feec522a9d 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala
@@ -171,7 +171,7 @@ class DataProcessor(
executor.close()
adaptiveBatchingMonitor.stopAdaptiveBatching()
stateManager.transitTo(COMPLETED)
- logger.info(
+ logger.debug(
s"$executor completed, # of input ports =
${inputManager.getAllPorts.size}, " +
s"input tuple count = ${statisticsManager.getInputTupleCount}, " +
s"output tuple count = ${statisticsManager.getOutputTupleCount}"
@@ -242,14 +242,14 @@ class DataProcessor(
): Unit = {
inputManager.currentChannelId = channelId
val command = ecm.commandMapping.get(actorId.name)
- logger.info(s"receive ECM from $channelId, id = ${ecm.id}, cmd = $command")
+ logger.debug(s"receive ECM from $channelId, id = ${ecm.id}, cmd =
$command")
if (ecm.ecmType != NO_ALIGNMENT) {
pauseManager.pauseInputChannel(ECMPause(ecm.id), List(channelId))
}
if (ecmManager.isECMAligned(channelId, ecm)) {
logManager.markAsReplayDestination(ecm.id)
// invoke the control command carried with the ECM
- logger.info(s"process ECM from $channelId, id = ${ecm.id}, cmd =
$command")
+ logger.debug(s"process ECM from $channelId, id = ${ecm.id}, cmd =
$command")
if (command.isDefined) {
// The reply must go back to the actor that originated the invocation
// (recorded in command.context.sender), not to channelId.fromWorkerId.
@@ -268,7 +268,7 @@ class DataProcessor(
outputManager.flush(Some(downstreamChannelsInScope))
outputGateway.getActiveChannels.foreach { activeChannelId =>
if (downstreamChannelsInScope.contains(activeChannelId)) {
- logger.info(
+ logger.debug(
s"send ECM to $activeChannelId, id = ${ecm.id}, cmd = $command"
)
outputGateway.sendTo(activeChannelId, ecm)
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala
index 5d1bf8ccbd..a9e3f4ed46 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/StartHandler.scala
@@ -39,7 +39,7 @@ trait StartHandler {
request: EmptyRequest,
ctx: AsyncRPCContext
): Future[WorkerStateResponse] = {
- logger.info("Starting the worker.")
+ logger.debug("Starting the worker.")
if (dp.executor.isInstanceOf[SourceOperatorExecutor]) {
val channelId =
ChannelIdentity(ActorVirtualIdentity("SOURCE_STARTER"), actorId,
isControl = false)
diff --git
a/amber/src/main/scala/org/apache/texera/amber/engine/common/client/ClientActor.scala
b/amber/src/main/scala/org/apache/texera/amber/engine/common/client/ClientActor.scala
index 6cff328296..9a2bc1ff62 100644
---
a/amber/src/main/scala/org/apache/texera/amber/engine/common/client/ClientActor.scala
+++
b/amber/src/main/scala/org/apache/texera/amber/engine/common/client/ClientActor.scala
@@ -156,6 +156,6 @@ private[client] class ClientActor extends Actor with
AmberLogging {
sender() ! Ack
coordinator ! x
case other =>
- logger.warn("client actor cannot handle " + other) //skip
+ logger.debug("client actor cannot handle " + other) //skip
}
}
diff --git a/common/config/src/main/resources/cluster.conf
b/common/config/src/main/resources/cluster.conf
index 524cd0216d..60feaa6ab4 100644
--- a/common/config/src/main/resources/cluster.conf
+++ b/common/config/src/main/resources/cluster.conf
@@ -23,6 +23,9 @@ pekko {
# Log level used by the configured loggers (see "loggers") as soon
# as they have been started; before that, see "stdout-loglevel"
# Options: OFF, ERROR, WARNING, INFO, DEBUG
+ # The env override may arrive in logback spelling (the same variable drives
+ # the logback root level in logback.xml);
PekkoConfig.normalizePekkoLogLevel
+ # translates e.g. WARN -> WARNING before this reaches any ActorSystem.
loglevel = "INFO"
loglevel = ${?TEXERA_SERVICE_LOG_LEVEL}
diff --git
a/common/config/src/main/scala/org/apache/texera/common/config/PekkoConfig.scala
b/common/config/src/main/scala/org/apache/texera/common/config/PekkoConfig.scala
index 449df5ef9d..d1159be70a 100644
---
a/common/config/src/main/scala/org/apache/texera/common/config/PekkoConfig.scala
+++
b/common/config/src/main/scala/org/apache/texera/common/config/PekkoConfig.scala
@@ -18,13 +18,37 @@
package org.apache.texera.common.config
-import com.typesafe.config.{Config, ConfigFactory}
+import com.typesafe.config.{Config, ConfigFactory, ConfigValueFactory}
object PekkoConfig {
// Load configuration
private val conf: Config =
ConfigFactory.parseResources("cluster.conf").resolve()
+ /**
+ * Translate a logback level spelling into one pekko accepts.
+ *
+ * cluster.conf forwards ${?TEXERA_SERVICE_LOG_LEVEL} into pekko.loglevel,
but that env
+ * var's vocabulary belongs to logback: the same variable drives the
logback root level
+ * in logback.xml, and logback cannot translate on its side (it silently
falls back to
+ * DEBUG on names it does not know, such as WARNING). Pekko in turn only
accepts OFF,
+ * ERROR, WARNING, INFO and DEBUG, and prints a LoggerException on every
ActorSystem
+ * creation before falling back to ERROR when handed a logback-only
spelling such as
+ * WARN. Translating here lets the single env knob drive both systems.
+ */
+ private[config] def normalizePekkoLogLevel(level: String): String =
+ level.toUpperCase match {
+ case "WARN" => "WARNING"
+ case "TRACE" | "ALL" => "DEBUG"
+ case other => other
+ }
+
// Return the complete Pekko configuration with fallback to default
application config
- def pekkoConfig: Config =
conf.withFallback(ConfigFactory.defaultApplication()).resolve()
+ def pekkoConfig: Config = {
+ val resolved =
conf.withFallback(ConfigFactory.defaultApplication()).resolve()
+ resolved.withValue(
+ "pekko.loglevel",
+
ConfigValueFactory.fromAnyRef(normalizePekkoLogLevel(resolved.getString("pekko.loglevel")))
+ )
+ }
}
diff --git
a/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala
b/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala
index 13282f286d..18dca80e87 100644
---
a/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala
+++
b/common/config/src/test/scala/org/apache/texera/common/config/PekkoConfigSpec.scala
@@ -76,4 +76,44 @@ class PekkoConfigSpec extends AnyFlatSpec with Matchers {
}
config.getString("pekko.stdout-loglevel") shouldBe "INFO"
}
+
+ it should "expose exactly the normalized form of whatever the env supplied"
in {
+ // Env-insensitive by construction: for ANY env value the exposed level
must
+ // equal its normalized form (which is the identity for spellings the
+ // translation doesn't know). This pins the plumbing — pekkoConfig actually
+ // routes pekko.loglevel through normalizePekkoLogLevel — and is exercised
+ // for real by CI, which sets the logback spelling WARN.
+ sys.env
+ .get("TEXERA_SERVICE_LOG_LEVEL")
+ .orElse(sys.props.get("TEXERA_SERVICE_LOG_LEVEL"))
+ .foreach { raw =>
+ PekkoConfig.pekkoConfig.getString("pekko.loglevel") shouldBe
+ PekkoConfig.normalizePekkoLogLevel(raw)
+ }
+ }
+
+ "PekkoConfig.normalizePekkoLogLevel" should "map known spellings into
pekko's set" in {
+ // Every spelling from the combined logback + pekko vocabulary must land in
+ // pekko's accepted set (Logging.levelFor); anything else makes every
+ // ActorSystem creation print a LoggerException and fall back to ERROR.
+ val pekkoAccepted = Set("OFF", "ERROR", "WARNING", "INFO", "DEBUG")
+ Seq("OFF", "ERROR", "WARN", "WARNING", "INFO", "DEBUG", "TRACE",
"ALL").foreach { spelling =>
+ pekkoAccepted should
contain(PekkoConfig.normalizePekkoLogLevel(spelling))
+ }
+ }
+
+ it should "translate logback-only spellings to pekko's" in {
+ PekkoConfig.normalizePekkoLogLevel("WARN") shouldBe "WARNING"
+ PekkoConfig.normalizePekkoLogLevel("warn") shouldBe "WARNING"
+ PekkoConfig.normalizePekkoLogLevel("TRACE") shouldBe "DEBUG"
+ PekkoConfig.normalizePekkoLogLevel("ALL") shouldBe "DEBUG"
+ }
+
+ it should "pass levels pekko already accepts through unchanged" in {
+ Seq("OFF", "ERROR", "WARNING", "INFO", "DEBUG").foreach { level =>
+ PekkoConfig.normalizePekkoLogLevel(level) shouldBe level
+ }
+ // pekko's levelFor is case-insensitive, so uppercasing valid input is
harmless
+ PekkoConfig.normalizePekkoLogLevel("info") shouldBe "INFO"
+ }
}
