HIVE-27126: queue level resource stats for YARN RM. - #6501
HIVE-27126: queue level resource stats for YARN RM.#6501architjainjain wants to merge 2 commits into
Conversation
ea3b1b3 to
5f4c748
Compare
asf-ci-hive
added
tests pending
tests failed
and removed
tests failed
tests pending
labels
9e97b27 to
e2d3173
Compare
asf-ci-hive
added
tests pending
tests passed
and removed
tests failed
tests pending
labels
e2d3173 to
fa844c0
Compare
fa844c0 to
dac58c2
Compare
asf-ci-hive
added
tests passed
tests pending
tests unstable
and removed
tests pending
tests passed
labels
dac58c2 to
f4301f1
Compare
f4301f1 to
ada32ce
Compare
asf-ci-hive
added
tests pending
tests unstable
and removed
tests passed
tests pending
labels
120c749 to
34febff
Compare
asf-ci-hive
added
tests pending
tests passed
and removed
tests unstable
tests pending
labels
34febff to
47b3abe
Compare
asf-ci-hive
added
tests pending
tests failed
and removed
tests passed
tests pending
labels
abstractdog
left a comment
abstractdog
left a comment
There was a problem hiding this comment.
thanks @architjainjain so far, dropped some comments, but I wasn't able to fully read it through, I'll get back after, in the meantime I can do some testing too hopefully
| * behaviour added as part of HIVE-27126. | ||
| * | ||
| * We capture stdout via a ByteArrayOutputStream and inspect the rendered output. | ||
| */public class TestInPlaceUpdate { |
There was a problem hiding this comment.
line break before public
There was a problem hiding this comment.
TestInPlaceUpdate disappeared in the meantime from the PR
if it's not needed anymore, feel free to resolve this comment, otherwise I'm happy to see that test case
There was a problem hiding this comment.
I have added back, test case.
| // HIVE-27126: Thrift regeneration omitted setStartTimeIsSet(true) from the constructor. | ||
| // Explicitly call setStartTime() to set the isset flag required for Thrift validation. | ||
| tProgressUpdateResp.setStartTime(progressUpdate.startTimeMillis); | ||
| if (progressUpdate.queueMetrics() != null && !progressUpdate.queueMetrics().isEmpty()) { | ||
| tProgressUpdateResp.setQueueMetrics(progressUpdate.queueMetrics()); | ||
| } | ||
| resp.setProgressUpdateResponse(tProgressUpdateResp); |
There was a problem hiding this comment.
why is this change needed?
There was a problem hiding this comment.
This works around a Thrift code generation bug. When we regenerated Thrift code after adding the queueMetrics field, the generated constructor for TProgressUpdateResp accepts startTimeMillis but fails to set the isset flag. Without this flag, Thrift serialization treats the field as unset, causing clients to receive incomplete progress updates. The explicit setStartTime() call ensures both the value AND the isset flag are properly set, maintaining backward compatibility with Thrift clients.
without this test the generated file is not having the starttime isset flag updated.
| this.status = status; | ||
| this.footerSummary = footerSummary; | ||
| this.startTime = startTime; | ||
| setStartTimeIsSet(true); |
There was a problem hiding this comment.
@abstractdog this is getting removed. so we added manually after the constructor call.
There was a problem hiding this comment.
this is not expected, see my other comment thrift code generation
https://github.com/apache/hive/pull/6501/changes#r3703255649
47b3abe to
e7a0726
Compare
There was a problem hiding this comment.
Pull request overview
Note
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Adds optional, real-time YARN queue resource metrics to the Tez progress display during Hive query execution, propagating the data through HS2/Thrift and Beeline in-place updates.
Changes:
- Introduces YARN queue metrics collection (collector + per-queue state + shared cache + scheduled refresh pool with jitter/circuit breaker).
- Integrates queue metrics into Tez progress rendering (in-place + log-to-file) and publishes it over Thrift (
TProgressUpdateResp). - Adds unit tests covering cache/state/pool behavior and Tez monitor formatting/initialization.
Reviewed changes
Copilot reviewed 29 out of 30 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| service/src/java/org/apache/hive/service/server/HiveServer2.java | Initializes the shared refresh pool during HS2 Tez session pool startup. |
| service/src/java/org/apache/hive/service/cli/thrift/ThriftCLIService.java | Populates new Thrift queueMetrics field and applies Thrift “isset” workaround for startTime. |
| service/src/java/org/apache/hive/service/cli/JobProgressUpdate.java | Adds queue metrics string to progress update model. |
| service-rpc/if/TCLIService.thrift | Adds optional queueMetrics to TProgressUpdateResp. |
| ql/src/java/org/apache/hadoop/hive/ql/session/SessionState.java | Implements queueMetrics() default for the progress monitor facade. |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/QueueMetricsCollector.java | New interface for queue metrics collectors. |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/NoOpQueueMetricsCollector.java | Null-object collector when feature is disabled. |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/QueueMetricsSnapshot.java | Immutable snapshot of queue resource stats fetched from RM. |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/QueueMetricsState.java | Per-queue state: interval registration, scheduling, refresh lock, circuit breaker. |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/QueueMetricsCache.java | JVM-wide Guava cache mapping queue name → state with expiry. |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/QueueMetricsRefreshPool.java | JVM-wide scheduled executor singleton to run refresh tasks (with jitter). |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/YarnQueueMetricsCollector.java | Active collector that coordinates with shared cache/state and refresh pool. |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/TezProgressMonitor.java | Formats and returns multi-line queue metrics for progress rendering. |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/TezJobMonitor.java | Creates/shuts down the metrics collector based on config and wires it into progress monitor. |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/RenderStrategy.java | Appends queue metrics into log-to-file progress report (single-line rendering). |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java | Adds per-session YarnClient lifecycle to support collector RM queries. |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolSession.java | Exposes YarnClient via pooled session wrapper. |
| ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSession.java | Extends TezSession interface with getYarnClient(). |
| common/src/java/org/apache/hadoop/hive/conf/HiveConf.java | Adds new configuration keys for refresh interval and refresh thread pool size. |
| common/src/java/org/apache/hadoop/hive/common/log/ProgressMonitor.java | Adds queueMetrics() to the ProgressMonitor contract. |
| common/src/java/org/apache/hadoop/hive/common/log/InPlaceUpdate.java | Renders queue metrics block (multi-line) under the standard progress output. |
| beeline/src/java/org/apache/hive/beeline/logs/BeelineInPlaceUpdateStream.java | Reads queueMetrics from Thrift progress updates for in-place rendering. |
| ql/src/test/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/TestQueueMetricsState.java | Unit tests for per-queue state logic (intervals, refresh lock, circuit breaker). |
| ql/src/test/org/apache/hadoop/hive/ql/exec/tez/monitoring/TestTezProgressMonitorQueueMetrics.java | Unit tests for queue metrics formatting and edge cases in TezProgressMonitor. |
| ql/src/test/org/apache/hadoop/hive/ql/exec/tez/monitoring/TestTezJobMonitorQueueMetrics.java | Unit tests for TezJobMonitor collector initialization decisions. |
| ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestYarnQueueMetricsCollector.java | Unit tests for collector/cache/pool integration behavior and circuit breaker. |
| ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestQueueMetricsRefreshPool.java | Unit tests for refresh pool singleton, scheduling, and jitter determinism/range. |
| ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestQueueMetricsCache.java | Unit tests for cache placeholder/put semantics and concurrency behavior. |
| ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestNoOpQueueMetricsCollector.java | Unit tests for NoOp collector behavior and singleton semantics. |
Files not reviewed (1)
- service-rpc/src/gen/thrift/gen-javabean/org/apache/hive/service/rpc/thrift/TProgressUpdateResp.java: Generated file
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
abstractdog
left a comment
abstractdog
left a comment
There was a problem hiding this comment.
thanks @architjainjain for the patch so far, left comments
| if (System.currentTimeMillis() - startTime > timeoutMs) { | ||
| fail("Snapshot not available after " + timeoutMs + "ms"); | ||
| } | ||
| Thread.onSpinWait(); // Hint to JVM that this is a spin-wait loop |
There was a problem hiding this comment.
I believe onSpinWait is just a sophisticated way to "busy wait", and in unit tests, there always must be a better way leveraging CountDownLatch, CyclicBarrier, this applies to other usages of Thread.onSpinWait
There was a problem hiding this comment.
replaced all Thread.onSpinWait() usages with Thread.sleep(10), which properly yields the CPU to other threads (including the background refresh thread) between each poll, rather than busy-waiting.
For the async scenarios (waitForSnapshot, waitForInvocationCount, post-shutdown stabilization), the polling loop with Thread.sleep(10) is the right approach since these results are driven by the background scheduled executor — there's no test-owned thread to coordinate with a CountDownLatch or CyclicBarrier without adding test hooks to production code.
CountDownLatch/CyclicBarrier are already used correctly in TestQueueMetricsCache for testConcurrentPutPlaceholderRaces and testConcurrentPutAndGetNoDeadlock, where the test itself creates and coordinates multiple threads — which is exactly the right use case for those primitives.
| if (System.currentTimeMillis() - startTime > timeoutMs) { | ||
| return; | ||
| } | ||
| Thread.onSpinWait(); // Hint to JVM that this is a spin-wait loop |
There was a problem hiding this comment.
| long deadline = System.currentTimeMillis() + 3000; | ||
| do { | ||
| callsAfterShutdown = mockingDetails(mockYarnClient).getInvocations().size(); | ||
| Thread.onSpinWait(); // Hint to JVM that this is a spin-wait loop |
There was a problem hiding this comment.
| // Spin-wait up to 200ms to ensure time has passed so the age comparison is meaningful | ||
| long deadline = System.currentTimeMillis() + 200; | ||
| while (state.getAgeMs() <= initialAge && System.currentTimeMillis() < deadline) { | ||
| Thread.onSpinWait(); // Hint to JVM that this is a spin-wait loop |
There was a problem hiding this comment.
| // Create multiple threads that call getYarnClient() concurrently | ||
| final int threadCount = 10; | ||
| Thread[] threads = new Thread[threadCount]; | ||
| YarnClient[] clients = new YarnClient[threadCount]; | ||
|
|
||
| for (int i = 0; i < threadCount; i++) { | ||
| final int index = i; | ||
| threads[i] = new Thread(() -> { | ||
| clients[index] = sessionState.getYarnClient(); | ||
| }); | ||
| } | ||
|
|
||
| // Start all threads | ||
| for (Thread thread : threads) { | ||
| thread.start(); | ||
| } | ||
|
|
||
| // Wait for all threads to complete | ||
| for (Thread thread : threads) { | ||
| thread.join(); | ||
| } | ||
|
|
||
| // All threads should get the same YarnClient instance | ||
| YarnClient firstClient = clients[0]; | ||
| Assert.assertNotNull("YarnClient should be initialized", firstClient); | ||
|
|
||
| for (int i = 1; i < threadCount; i++) { | ||
| Assert.assertSame("All threads should get the same YarnClient instance", firstClient, clients[i]); | ||
| } | ||
| } |
There was a problem hiding this comment.
I don't think we have to write "thread creation and wait" from scratch, what about something like this:
int threadCount = 10;
ExecutorService executor = Executors.newFixedThreadPool(threadCount);
CountDownLatch start = new CountDownLatch(1);
try {
List<Future<YarnClient>> futures = IntStream.range(0, threadCount)
.mapToObj(i -> executor.submit(() -> {
start.await();
return sessionState.getYarnClient();
}))
.collect(Collectors.toList());
start.countDown();
YarnClient firstClient = futures.get(0).get();
assertNotNull("YarnClient should be initialized", firstClient);
for (Future<YarnClient> future : futures) {
assertSame(
"All threads should get the same YarnClient instance",
firstClient,
future.get());
}
} finally {
executor.shutdownNow();
}
CountDownLatch also helps testing high concurrency
| this.status = status; | ||
| this.footerSummary = footerSummary; | ||
| this.startTime = startTime; | ||
| setStartTimeIsSet(true); |
There was a problem hiding this comment.
this is not expected, see my other comment thrift code generation
https://github.com/apache/hive/pull/6501/changes#r3703255649
abstractdog
left a comment
abstractdog
left a comment
There was a problem hiding this comment.
thanks @architjainjain for the patch so far, left comments
I will look into this and provide fix for the review comments added. |
| * queue metrics is controlled per-session by {@code hive.tez.queue.metrics.refresh.interval}, | ||
| * which is checked when creating metrics collectors for each query. | ||
| * <p> | ||
| * In non-Tez environments (MR, Spark, local), the pool is not created, avoiding unnecessary |
| @@ -0,0 +1,247 @@ | |||
| /* | |||
There was a problem hiding this comment.
@abstractdog This is the new test case added
architjainjain
left a comment
architjainjain
left a comment
There was a problem hiding this comment.
@abstractdog All commnets are addressed and updated/added/removed test case as per the changes.
Thanks for the review.
|
| private static final int DEFAULT_THREAD_COUNT = 4; | ||
| public static final int JITTER_PERCENT = 10; | ||
|
|
||
| private static final AtomicReference<QueueMetricsRefreshPool> instance = new AtomicReference<>(null); |
There was a problem hiding this comment.
@abstractdog you comment was regarding usage of AtomicReference, and only volatile can be sufficient.
Switched to AtomicReference because:
volatile alone isn't thread-safe for compound operations - The check-then-act pattern (if (instance == null) instance = new...) requires synchronization
AtomicReference provides atomic compare-and-set - Single atomic operation vs. separate check + lock + set
Lock-free - No thread blocking, better for concurrent HiveServer2 startup
Explicit concurrency handling - compareAndSet() makes race condition handling visible in code
The original volatile + synchronized was correct but required locks. AtomicReference achieves thread-safety without blocking.
What changes were proposed in this pull request?
This PR adds real-time YARN queue resource metrics display alongside Tez job progress during query execution. Users can now see memory, vCores, capacity utilization, running/pending applications, and allocated/pending containers for the queue being used by their queries.
Key Components:
What is NOT Changed:
Technical Highlights:
Configuration:
hive.tez.queue.metrics.refresh.interval: Per-session refresh interval (default: 0s)hive.server2.tez.queue.metrics.refresh.threads: Pool size for metric collection (default: 4)Important Note: An HTTP exporter and Prometheus/Grafana integration were implemented as a local testing setup to validate the feature during development. These components are NOT part of this PR, NOT included in production code, and were only used in a local Docker environment for validation purposes.
Why are the changes needed?
Currently, when a Hive query is slow or stalled, users cannot determine if the issue is due to insufficient queue resources or other factors. They see Tez task progress (e.g., "Map 1: 3/10") but have no visibility into whether the queue has available memory/vCores to execute tasks in parallel.
This enhancement provides real-time queue resource information, enabling users to:
Backward Compatibility:
Performance Impact:
Does this PR introduce any user-facing change?
Yes. When
hive.tez.queue.metrics.refresh.intervalis set to a positive value (default: 0s), users will see queue-level metrics displayed with Tez job progress:In-place mode (hive.server2.in.place.progress=true):
Log file mode (hive.server2.in.place.progress=false):
When disabled (set hive.tez.queue.metrics.refresh.interval=0s):
No queue metrics are displayed, behavior remains identical to previous versions.
How was this patch tested?
The patch was tested in a multi-node YARN cluster environment with concurrent queries running on different queues.
Testing included:
Test Environment:
Validation artifacts (local testing setup only - NOT in production):
Screenshots:
1. Terminal output with queue metrics enabled (hive.tez.queue.metrics.refresh.interval=10s):
2. Terminal output with queue metrics disabled (hive.tez.queue.metrics.refresh.interval=0s):
3. Grafana dashboard showing backend metrics validation (local testing setup only):
Note: The Grafana and Prometheus infrastructure shown above was created exclusively for local testing and validation purposes. This monitoring stack is NOT part of the production code, NOT included in the deployment, and was only used in a local Docker environment to validate the feature's correctness.
Configuration tested: