j1wonpark opened a new pull request, #4197:
URL: https://github.com/apache/amoro/pull/4197
## Why are the changes needed?
When an optimizer process receives SIGTERM (e.g. K8s pod termination, manual
rolling
restart), in-progress tasks should be allowed to finish and report their
results to
AMS instead of being terminated mid-execution.
Today `Optimizer.stopOptimizing()` only flips the stopped flag and returns
immediately, and any subsequent `OptimizerExecutor.completeTask()` call is
gated by
`isStarted()` inside `callAuthenticatedAms`, so the in-flight result is
silently
dropped. The same task is later re-scheduled by AMS, doubling work and
producing
duplicate commits when the original task was already past its commit point.
In containerized deployments, several signal-delivery and shutdown-ordering
issues
prevent the graceful path from running even when the code supports it:
1. The optimizer container is launched with `sh -c <args>`, which leaves
`sh` as
PID 1. SIGTERM is delivered to `sh` and not forwarded to the java child,
so the
pod is killed by SIGKILL after `terminationGracePeriodSeconds` elapses.
2. The default K8s `terminationGracePeriodSeconds` (30s) is too short for
in-flight
optimizing tasks (compaction, sorting) to drain.
3. Hadoop's FileSystem cache cleanup hook runs concurrently with JVM Runtime
shutdown hooks, so an in-flight HDFS writer can hit
`ClosedChannelException`
during row-group flush while our hook is still waiting for the task to
complete.
## Brief change log
**Core shutdown flow**
- `Optimizer.stopOptimizing()`: join executor threads with a configurable
deadline;
force-interrupt + 1s join only on timeout. Stop executors first but keep
the
toucher alive so AMS heartbeats continue while in-flight tasks drain —
otherwise
AMS hits its heartbeat-timeout and unregisters the optimizer, causing the
best-effort `completeTask` to fail with "not authenticated".
- `OptimizerExecutor.completeTask()`: when the operator is already stopped,
fall
back to a single best-effort direct call so AMS still receives the
completion
instead of having the result silently dropped by the gated retry loop.
- `OptimizerToucher.stop()`: interrupt the runner thread to wake it from
`waitAShortTime` so the heartbeat loop terminates promptly.
- `AbstractOptimizerOperator.waitAShortTime()`: preserve the interrupt flag
so
upper loops observe the interrupt instead of swallowing it.
- `OptimizerConfig`: new `-st` / `--shutdown-timeout-ms` option, default
600000ms
(10 min). Operators should align this with their cluster termination grace
period.
**Signal delivery and shutdown ordering**
- `SparkOptimizer` / `StandaloneOptimizer`: register graceful shutdown on
Hadoop's
`ShutdownHookManager` with a priority above `FS_CACHE` (and
`SPARK_CONTEXT_SHUTDOWN_PRIORITY` for Spark) so our hook completes before
the
FileSystem cache and SparkContext are torn down. An explicit per-hook
timeout is
passed to override the `hadoop.service.shutdown.timeout` default of 30s.
- `KubernetesOptimizerContainer`: container command wrapped with
`sh -c 'exec <args>'` so the entrypoint replaces the shell and java
receives
SIGTERM directly. `terminationGracePeriodSeconds` is set on the generated
PodSpec, derived from the optimizer's shutdown-timeout-ms plus a 30s
buffer for
thread join, result reporting, and log flush. A user-supplied podTemplate
value,
if present, is respected.
- `optimizer.sh start-foreground`: `exec $CMDS` so java takes PID 1 and
receives
SIGTERM directly when run from the standalone bin.
## How was this patch tested?
- [x] Add some test cases that check the changes thoroughly including
negative and positive cases if possible
- `TestOptimizer.testGracefulShutdownWaitsForInProgressTask` —
long-running task
started before `stopOptimizing()` runs to completion and SUCCESS is
reported to
AMS, with an assert that ensures `stopOptimizing` actually blocked.
- `TestOptimizer.testGracefulShutdown` — basic graceful shutdown still
reports
completed task results.
- `TestOptimizingInput.slowSuccessInput` — the test executor sleeps for the
requested duration and surfaces an error if the thread is interrupted,
so a
regression that interrupts the in-flight task fails the assertion.
- `TestKubernetesOptimizerContainer` — five new tests covering the `exec`
prefix
in the container command (with and without podTemplate), default
`terminationGracePeriodSeconds` derivation, `-st` arg derivation, and
respect
for a user-supplied podTemplate value.
- [x] Run test locally before making a pull request
- `./mvnw test -pl amoro-optimizer/amoro-optimizer-common` → **12 / 12
PASS**
- `./mvnw test -pl amoro-ams -am -Dtest=TestKubernetesOptimizerContainer` →
**9 / 9 PASS** (5 new + 4 existing)
- [ ] Add screenshots for manual tests if appropriate (N/A — runtime
behavior)
## Documentation
- Does this pull request introduce a new feature? **yes** (new `-st` /
`--shutdown-timeout-ms` option, new K8s `terminationGracePeriodSeconds`
derivation)
- If yes, how is the feature documented? **JavaDocs / option `usage`
string** —
the option's `usage` string in `OptimizerConfig` describes intent and the
alignment with K8s `terminationGracePeriodSeconds`. User-facing docs update
recommended in a follow-up if maintainers prefer.
--
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]
