Copilot commented on code in PR #6501:
URL: https://github.com/apache/hive/pull/6501#discussion_r3656958152
##########
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java:
##########
@@ -797,6 +810,19 @@ public String getSessionId() {
protected final void setTezClient(TezClient session) {
this.session = session;
+
+ // Initialize YarnClient for queue metrics collection
+ if (session != null && yarnClient == null) {
+ try {
+ yarnClient = YarnClient.createYarnClient();
+ yarnClient.init(conf);
+ yarnClient.start();
+ LOG.info("YarnClient initialized for session: {}", sessionId);
+ } catch (Exception e) {
+ LOG.warn("Failed to initialize YarnClient for metrics collection", e);
+ yarnClient = null;
+ }
+ }
Review Comment:
YarnClient is initialized for every Tez session as soon as TezClient is set,
even when queue metrics are disabled (interval=0). This creates extra
threads/connections and contradicts the intended ‘zero impact when disabled’.
Consider lazy-initializing YarnClient only when metrics are enabled (e.g., gate
on `HIVE_TEZ_QUEUE_METRICS_REFRESH_INTERVAL` or initialize on first
`getYarnClient()` call) so disabled configs don’t incur YARN client overhead.
##########
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/QueueMetricsRefreshPool.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.hadoop.hive.ql.exec.tez.monitoring.yarnqueue;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Singleton manager for the JVM-wide refresh executor pool used by queue
metrics collection.
+ * Provides a shared {@link ScheduledExecutorService} that fires periodic YARN
RM refresh tasks
+ * across all queries in the HiveServer2 process.
+ * <p>
+ * Initialized during HiveServer2 startup via {@link #init(int)} when Tez
session pool is set up.
+ * Pool size is configured via {@code
hive.server2.tez.queue.metrics.refresh.threads} (default: 4).
+ * <p>
+ * All {@link YarnQueueMetricsCollector} instances share this single pool,
ensuring efficient
+ * resource usage and preventing thread explosion when many queries run
concurrently.
+ * <p>
+ * Thread-safe singleton implementation using double-check locking pattern.
+ */
+public final class QueueMetricsRefreshPool {
+ private static final Logger LOG =
LoggerFactory.getLogger(QueueMetricsRefreshPool.class);
+
+ private static final int DEFAULT_THREAD_COUNT = 4;
+ public static final int JITTER_PERCENT = 10;
+
+ private static final AtomicReference<QueueMetricsRefreshPool> INSTANCE = new
AtomicReference<>(null);
+ private static final Object INIT_LOCK = new Object();
+
+ private final ScheduledExecutorService refreshPool;
+
+
+ /**
+ * Initializes the singleton pool with the specified thread count.
+ * Must be called during HiveServer2 startup. Subsequent calls are ignored.
+ *
+ * @param threadCount number of threads for the refresh pool
+ */
+ public static void init(int threadCount) {
+ if (INSTANCE.get() != null) {
+ LOG.debug("QueueMetricsRefreshPool already initialized, ignoring init
call");
+ return;
+ }
+ synchronized (INIT_LOCK) {
+ if (INSTANCE.get() == null) {
+ INSTANCE.set(new QueueMetricsRefreshPool(threadCount));
+ }
+ }
+ }
+
+ /**
+ * Returns the singleton instance. Must be called after {@link #init(int)}.
+ * For tests or non-HS2 environments, lazily initializes with default thread
count.
+ *
+ * @return the singleton pool instance
+ */
+ public static QueueMetricsRefreshPool getInstance() {
+ QueueMetricsRefreshPool local = INSTANCE.get();
+ if (local != null) {
+ return local;
+ }
+ // Lazy init for tests/non-HS2 with default thread count
+ synchronized (INIT_LOCK) {
+ local = INSTANCE.get();
+ if (local == null) {
+ LOG.warn("QueueMetricsRefreshPool not initialized via init(), using
default thread count: {}",
+ DEFAULT_THREAD_COUNT);
+ local = new QueueMetricsRefreshPool(DEFAULT_THREAD_COUNT);
+ INSTANCE.set(local);
+ }
+ return local;
+ }
+ }
+
+
+ private QueueMetricsRefreshPool(int threadCount) {
+ this.refreshPool = Executors.newScheduledThreadPool(threadCount,
+ new ThreadFactoryBuilder()
+ .setNameFormat("queue-metrics-refresh-%d")
+ .setDaemon(true)
+ .build());
+ LOG.info("QueueMetricsRefreshPool initialized with {} threads",
threadCount);
+ }
+
+
+ /**
+ * Schedules a periodic refresh task. initialDelay=0 so the first fetch runs
immediately.
+ */
+ public ScheduledFuture<Void> scheduleRefreshTask(Runnable task, long
intervalMs) {
+ return (ScheduledFuture<Void>) refreshPool.scheduleWithFixedDelay(task, 0,
intervalMs, TimeUnit.MILLISECONDS);
+ }
Review Comment:
This method performs an unchecked cast from `ScheduledFuture<?>` to
`ScheduledFuture<Void>`. Consider changing the return type to
`ScheduledFuture<?>` (or `ScheduledFuture<?>` consistently across call sites)
and removing the cast to avoid unsafe generics and potential warnings/future
incompatibilities.
##########
service/src/java/org/apache/hive/service/server/HiveServer2.java:
##########
@@ -966,6 +970,20 @@ private void initAndStartWorkloadManager(final
WMFullResourcePlan resourcePlan)
}
}
+ /**
+ * Initializes the queue metrics refresh pool and HTTP exporter.
+ * Failures are non-fatal — logged as warnings so the server can start
without queue metrics.
+ */
+ private void initializeQueueMetricsPool(HiveConf hiveConf) {
+ try {
+ int refreshThreads =
hiveConf.getIntVar(ConfVars.HIVE_SERVER2_TEZ_QUEUE_METRICS_REFRESH_THREADS);
+ QueueMetricsRefreshPool.init(refreshThreads);
+ LOG.info("Queue metrics refresh pool initialized with {} threads",
refreshThreads);
+ } catch (Exception e) {
+ LOG.warn("Failed to initialize queue metrics refresh pool: {}",
e.getMessage());
+ }
Review Comment:
The warning drops the stack trace by logging only `e.getMessage()`, which
makes diagnosing startup issues much harder in production. Log the exception
object as well (e.g., `LOG.warn(\"...\", e)`), or include both message and
throwable.
##########
service/src/java/org/apache/hive/service/server/HiveServer2.java:
##########
@@ -966,6 +970,20 @@ private void initAndStartWorkloadManager(final
WMFullResourcePlan resourcePlan)
}
}
+ /**
+ * Initializes the queue metrics refresh pool and HTTP exporter.
+ * Failures are non-fatal — logged as warnings so the server can start
without queue metrics.
+ */
+ private void initializeQueueMetricsPool(HiveConf hiveConf) {
Review Comment:
The method comment mentions initializing an “HTTP exporter”, but the
implementation only initializes `QueueMetricsRefreshPool`. This is misleading,
and it also conflicts with the PR description stating the HTTP exporter is not
part of this PR. Update the Javadoc to reflect what is actually initialized
here.
##########
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/QueueMetricsRefreshPool.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.hadoop.hive.ql.exec.tez.monitoring.yarnqueue;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Singleton manager for the JVM-wide refresh executor pool used by queue
metrics collection.
+ * Provides a shared {@link ScheduledExecutorService} that fires periodic YARN
RM refresh tasks
+ * across all queries in the HiveServer2 process.
+ * <p>
+ * Initialized during HiveServer2 startup via {@link #init(int)} when Tez
session pool is set up.
+ * Pool size is configured via {@code
hive.server2.tez.queue.metrics.refresh.threads} (default: 4).
+ * <p>
+ * All {@link YarnQueueMetricsCollector} instances share this single pool,
ensuring efficient
+ * resource usage and preventing thread explosion when many queries run
concurrently.
+ * <p>
+ * Thread-safe singleton implementation using double-check locking pattern.
+ */
+public final class QueueMetricsRefreshPool {
+ private static final Logger LOG =
LoggerFactory.getLogger(QueueMetricsRefreshPool.class);
+
+ private static final int DEFAULT_THREAD_COUNT = 4;
+ public static final int JITTER_PERCENT = 10;
+
+ private static final AtomicReference<QueueMetricsRefreshPool> INSTANCE = new
AtomicReference<>(null);
+ private static final Object INIT_LOCK = new Object();
+
+ private final ScheduledExecutorService refreshPool;
+
+
+ /**
+ * Initializes the singleton pool with the specified thread count.
+ * Must be called during HiveServer2 startup. Subsequent calls are ignored.
+ *
+ * @param threadCount number of threads for the refresh pool
+ */
+ public static void init(int threadCount) {
+ if (INSTANCE.get() != null) {
+ LOG.debug("QueueMetricsRefreshPool already initialized, ignoring init
call");
+ return;
+ }
+ synchronized (INIT_LOCK) {
+ if (INSTANCE.get() == null) {
+ INSTANCE.set(new QueueMetricsRefreshPool(threadCount));
+ }
+ }
+ }
+
+ /**
+ * Returns the singleton instance. Must be called after {@link #init(int)}.
+ * For tests or non-HS2 environments, lazily initializes with default thread
count.
+ *
+ * @return the singleton pool instance
+ */
+ public static QueueMetricsRefreshPool getInstance() {
+ QueueMetricsRefreshPool local = INSTANCE.get();
+ if (local != null) {
+ return local;
+ }
+ // Lazy init for tests/non-HS2 with default thread count
+ synchronized (INIT_LOCK) {
+ local = INSTANCE.get();
+ if (local == null) {
+ LOG.warn("QueueMetricsRefreshPool not initialized via init(), using
default thread count: {}",
+ DEFAULT_THREAD_COUNT);
+ local = new QueueMetricsRefreshPool(DEFAULT_THREAD_COUNT);
+ INSTANCE.set(local);
+ }
+ return local;
+ }
+ }
+
+
+ private QueueMetricsRefreshPool(int threadCount) {
+ this.refreshPool = Executors.newScheduledThreadPool(threadCount,
+ new ThreadFactoryBuilder()
+ .setNameFormat("queue-metrics-refresh-%d")
+ .setDaemon(true)
+ .build());
+ LOG.info("QueueMetricsRefreshPool initialized with {} threads",
threadCount);
+ }
+
+
+ /**
+ * Schedules a periodic refresh task. initialDelay=0 so the first fetch runs
immediately.
+ */
+ public ScheduledFuture<Void> scheduleRefreshTask(Runnable task, long
intervalMs) {
+ return (ScheduledFuture<Void>) refreshPool.scheduleWithFixedDelay(task, 0,
intervalMs, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Calculates deterministic hash-based jitter for a queue name to prevent
thundering herd.
+ * Jitter range: 0 to (intervalMs * JITTER_PERCENT / 100)
+ *
+ * @param queueName the queue name to hash
+ * @param intervalMs the refresh interval in milliseconds
+ * @return jitter value in milliseconds (0 to intervalMs * 10%)
+ */
+ public static long calculateJitter(String queueName, long intervalMs) {
+ long jitterWindow = intervalMs * JITTER_PERCENT / 100;
+ return Math.abs(queueName.hashCode()) % jitterWindow;
+ }
Review Comment:
`calculateJitter` can throw `ArithmeticException` when `jitterWindow`
becomes 0 (e.g., very small intervals), and it can return a negative value when
`queueName.hashCode()` is `Integer.MIN_VALUE` because
`Math.abs(Integer.MIN_VALUE)` remains negative. Guard `jitterWindow <= 0` by
returning 0, and use a non-negative hash conversion (e.g.,
`queueName.hashCode() & 0x7fffffff` or `Integer.toUnsignedLong`) to guarantee
jitter is always in [0, jitterWindow).
##########
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/QueueMetricsRefreshPool.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.hadoop.hive.ql.exec.tez.monitoring.yarnqueue;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Singleton manager for the JVM-wide refresh executor pool used by queue
metrics collection.
+ * Provides a shared {@link ScheduledExecutorService} that fires periodic YARN
RM refresh tasks
+ * across all queries in the HiveServer2 process.
+ * <p>
+ * Initialized during HiveServer2 startup via {@link #init(int)} when Tez
session pool is set up.
+ * Pool size is configured via {@code
hive.server2.tez.queue.metrics.refresh.threads} (default: 4).
+ * <p>
+ * All {@link YarnQueueMetricsCollector} instances share this single pool,
ensuring efficient
+ * resource usage and preventing thread explosion when many queries run
concurrently.
+ * <p>
+ * Thread-safe singleton implementation using double-check locking pattern.
+ */
+public final class QueueMetricsRefreshPool {
+ private static final Logger LOG =
LoggerFactory.getLogger(QueueMetricsRefreshPool.class);
+
+ private static final int DEFAULT_THREAD_COUNT = 4;
+ public static final int JITTER_PERCENT = 10;
Review Comment:
The PR description states jitter is ‘0-20%’, but the implementation uses
`JITTER_PERCENT = 10` (i.e., 0-10%). Either update the PR description to match
the code, or adjust the constant/implementation to deliver the documented
jitter range.
##########
ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestYarnQueueMetricsCollector.java:
##########
@@ -0,0 +1,588 @@
+/*
+ * 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.hadoop.hive.ql.exec.tez;
+
+import
org.apache.hadoop.hive.ql.exec.tez.monitoring.yarnqueue.QueueMetricsCache;
+import
org.apache.hadoop.hive.ql.exec.tez.monitoring.yarnqueue.QueueMetricsRefreshPool;
+import
org.apache.hadoop.hive.ql.exec.tez.monitoring.yarnqueue.QueueMetricsSnapshot;
+import
org.apache.hadoop.hive.ql.exec.tez.monitoring.yarnqueue.YarnQueueMetricsCollector;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.yarn.api.records.QueueInfo;
+import org.apache.hadoop.yarn.api.records.QueueStatistics;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mockingDetails;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test cases for YarnQueueMetricsCollector.
+ */
+public class TestYarnQueueMetricsCollector {
+
+ @Mock
+ private YarnClient mockYarnClient;
+
+ @Mock
+ private QueueInfo mockQueueInfo;
+
+ @Mock
+ private QueueStatistics mockQueueStats;
+
+ private AutoCloseable closeable;
+ private HiveConf testConf;
+
+ private static final long WAIT_TIMEOUT_MS = 5000;
+
+ @Before
+ public void setUp() {
+ closeable = MockitoAnnotations.openMocks(this);
+ testConf = new HiveConf();
+ // Reset the pool manager singleton and cache so each test starts with a
clean state.
+ QueueMetricsRefreshPool.resetForTesting();
+ QueueMetricsCache.resetForTesting();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (closeable != null) {
+ closeable.close();
+ }
+ QueueMetricsRefreshPool.resetForTesting();
+ QueueMetricsCache.resetForTesting();
+ }
+
+ /**
+ * Helper to create a collector in tests using a default HiveConf (min pool
sizes).
+ */
+ private YarnQueueMetricsCollector newCollector(YarnClient yarnClient, String
queueName,
+ long refreshIntervalMs, String queryId) {
+ return new YarnQueueMetricsCollector(yarnClient, queueName,
refreshIntervalMs, queryId, testConf);
+ }
+
+ /**
+ * Waits for a snapshot to be available (non-null).
+ */
+ private QueueMetricsSnapshot waitForSnapshot(
+ YarnQueueMetricsCollector collector, long timeoutMs) {
+ long startTime = System.currentTimeMillis();
+ QueueMetricsSnapshot snapshot;
+ while ((snapshot = collector.getLatestSnapshot()) == null) {
+ if (System.currentTimeMillis() - startTime > timeoutMs) {
+ fail("Snapshot not available after " + timeoutMs + "ms");
+ }
+ }
+ return snapshot;
Review Comment:
This is a tight busy-wait loop that can burn CPU for up to the full timeout
and make the test suite noisy/flaky under load. Add a small backoff (e.g.,
`Thread.sleep(5-20ms)`), `LockSupport.parkNanos`, or use an awaiting utility so
the loop yields while waiting for async refresh.
##########
ql/src/test/org/apache/hadoop/hive/ql/exec/tez/monitoring/TestTezJobMonitorQueueMetrics.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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.hadoop.hive.ql.exec.tez.monitoring;
+
+import
org.apache.hadoop.hive.ql.exec.tez.monitoring.yarnqueue.NoOpQueueMetricsCollector;
+import
org.apache.hadoop.hive.ql.exec.tez.monitoring.yarnqueue.QueueMetricsCollector;
+import
org.apache.hadoop.hive.ql.exec.tez.monitoring.yarnqueue.YarnQueueMetricsCollector;
+
+import java.lang.reflect.Field;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConfForTest;
+import org.apache.hadoop.hive.ql.Context;
+import org.apache.hadoop.hive.ql.exec.tez.TezSession;
+import org.apache.hadoop.hive.ql.log.PerfLogger;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.session.SessionState;
+import org.apache.hadoop.yarn.client.api.YarnClient;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.DAG;
+import org.apache.tez.dag.api.client.DAGClient;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test cases for TezJobMonitor queue metrics initialization.
+ */
+public class TestTezJobMonitorQueueMetrics {
+
+ @Mock
+ private TezSession mockSession;
+ @Mock
+ private DAGClient mockDagClient;
+ @Mock
+ private DAG mockDag;
+ @Mock
+ private Context mockContext;
+ @Mock
+ private PerfLogger mockPerfLogger;
+ @Mock
+ private YarnClient mockYarnClient;
+ @Mock
+ private TezCounters mockCounters;
+
+ private HiveConf hiveConf;
+ private List<BaseWork> topSortedWorks;
+ private SessionState sessionState;
+
+ @Before
+ public void setUp() {
+ MockitoAnnotations.openMocks(this);
+ hiveConf = new HiveConfForTest(TestTezJobMonitorQueueMetrics.class);
+ hiveConf.set("hive.security.authorization.manager",
+
"org.apache.hadoop.hive.ql.security.authorization.plugin.sqlstd.SQLStdConfOnlyAuthorizerFactory");
+ sessionState = SessionState.start(hiveConf);
+ topSortedWorks = new ArrayList<>();
+ when(mockDag.getName()).thenReturn("test-dag-1");
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (sessionState != null) {
+ sessionState.close();
+ }
+ }
+
+ @Test
+ public void testMetricsCollectorDisabledByDefault() throws Exception {
+ when(mockSession.getYarnClient()).thenReturn(mockYarnClient);
+ when(mockSession.getQueueName()).thenReturn("default");
+
+ TezJobMonitor monitor =
+ new TezJobMonitor(mockSession, topSortedWorks, mockDagClient,
hiveConf, mockDag, mockContext, mockCounters,
+ mockPerfLogger);
+
+ assertNotNull("Monitor should be created", monitor);
+ verify(mockSession, never()).getYarnClient();
+ verify(mockYarnClient, never()).getQueueInfo(anyString());
+ }
+
+ @Test
+ public void testMetricsCollectorEnabledWithInterval() {
+
hiveConf.setTimeVar(HiveConf.ConfVars.HIVE_TEZ_QUEUE_METRICS_REFRESH_INTERVAL,
10, TimeUnit.SECONDS);
+
+ when(mockSession.getYarnClient()).thenReturn(mockYarnClient);
+ when(mockSession.getQueueName()).thenReturn("default");
+
+ TezJobMonitor monitor =
+ new TezJobMonitor(mockSession, topSortedWorks, mockDagClient,
hiveConf, mockDag, mockContext, mockCounters,
+ mockPerfLogger);
+
+ assertNotNull("Monitor should be created", monitor);
+ verify(mockSession, atLeastOnce()).getYarnClient();
+ }
+
+ @Test
+ public void testMetricsCollectorDisabledWithZeroInterval() throws Exception {
+
hiveConf.setTimeVar(HiveConf.ConfVars.HIVE_TEZ_QUEUE_METRICS_REFRESH_INTERVAL,
0, TimeUnit.SECONDS);
+
+ when(mockSession.getYarnClient()).thenReturn(mockYarnClient);
+ when(mockSession.getQueueName()).thenReturn("default");
+
+ TezJobMonitor monitor =
+ new TezJobMonitor(mockSession, topSortedWorks, mockDagClient,
hiveConf, mockDag, mockContext, mockCounters,
+ mockPerfLogger);
+
+ assertNotNull("Monitor should be created", monitor);
+ verify(mockSession, never()).getYarnClient();
+ verify(mockYarnClient, never()).getQueueInfo(anyString());
+ }
+
+ @Test
+ public void testMetricsCollectorDisabledWithNegativeInterval() throws
Exception {
+
hiveConf.setTimeVar(HiveConf.ConfVars.HIVE_TEZ_QUEUE_METRICS_REFRESH_INTERVAL,
-1, TimeUnit.SECONDS);
+
+ when(mockSession.getYarnClient()).thenReturn(mockYarnClient);
+ when(mockSession.getQueueName()).thenReturn("default");
+
+ TezJobMonitor monitor =
+ new TezJobMonitor(mockSession, topSortedWorks, mockDagClient,
hiveConf, mockDag, mockContext, mockCounters,
+ mockPerfLogger);
+
+ assertNotNull("Monitor should be created", monitor);
+ verify(mockSession, never()).getYarnClient();
+ verify(mockYarnClient, never()).getQueueInfo(anyString());
+ }
+
+ @Test
+ public void testMetricsCollectorWithSmallInterval() {
+
hiveConf.setTimeVar(HiveConf.ConfVars.HIVE_TEZ_QUEUE_METRICS_REFRESH_INTERVAL,
500, TimeUnit.MILLISECONDS);
+
+ when(mockSession.getYarnClient()).thenReturn(mockYarnClient);
+ when(mockSession.getQueueName()).thenReturn("default");
+
+ TezJobMonitor monitor =
+ new TezJobMonitor(mockSession, topSortedWorks, mockDagClient,
hiveConf, mockDag, mockContext, mockCounters,
+ mockPerfLogger);
+
+ assertNotNull("Monitor should be created with adjusted interval", monitor);
+ }
+
+ @Test
+ public void testMetricsCollectorWithCustomQueue() {
+
hiveConf.setTimeVar(HiveConf.ConfVars.HIVE_TEZ_QUEUE_METRICS_REFRESH_INTERVAL,
15, TimeUnit.SECONDS);
+
+ when(mockSession.getYarnClient()).thenReturn(mockYarnClient);
+ when(mockSession.getQueueName()).thenReturn("production.analytics");
+
+ TezJobMonitor monitor =
+ new TezJobMonitor(mockSession, topSortedWorks, mockDagClient,
hiveConf, mockDag, mockContext, mockCounters,
+ mockPerfLogger);
+
+ verify(mockSession, atLeastOnce()).getQueueName();
+ assertNotNull("Monitor should be created with custom queue", monitor);
+ }
+
+ /**
+ * Tests that TezJobMonitor handles edge cases gracefully when metrics
collection
+ * cannot be initialized due to null or invalid inputs.
+ * <p>
+ * This test covers three edge cases in a single parameterized test to avoid
+ * code duplication (SonarQube rule java:S5976):
+ * <ul>
+ * <li>nullYarnClient: YarnClient is null</li>
+ * <li>nullQueueName: Queue name is null</li>
+ * <li>blankQueueName: Queue name is blank/whitespace</li>
+ * </ul>
+ * <p>
+ * All three cases share identical structure (metrics enabled with 10s
interval,
+ * one edge-case input), so they are expressed as a single test with an
inline
+ * parameter table.
+ *
+ * <pre>
+ * label | yarnClient | getQueueName()
+ * nullYarnClient | null | "default"
+ * nullQueueName | mock | null
+ * blankQueueName | mock | " "
+ * </pre>
+ */
+ @Test
+ public void testMetricsCollectorWithEdgeCaseQueueNameOrYarnClient() {
+ Object[][] cases = {
+ {"nullYarnClient", null, "default"},
+ {"nullQueueName", mockYarnClient, null},
+ {"blankQueueName", mockYarnClient, " "}
+ };
+
+ for (Object[] row : cases) {
+ String label = (String) row[0];
+ YarnClient yarnClient = (YarnClient) row[1];
+ String queueName = (String) row[2];
+
+ // Re-initialise mocks so state from the previous row does not leak.
+ MockitoAnnotations.openMocks(this);
+ when(mockDag.getName()).thenReturn("test-dag-1");
Review Comment:
Calling `MockitoAnnotations.openMocks(this)` inside the loop reinitializes
mocks repeatedly without closing the previous `AutoCloseable`, which can leak
resources and makes the test harder to reason about. Prefer restructuring as a
proper parameterized test (JUnit runner) or extract a helper that constructs
fresh local mocks per case (and closes them), avoiding repeated global re-init
within the same test method.
##########
common/src/java/org/apache/hadoop/hive/conf/HiveConf.java:
##########
@@ -4006,6 +4006,15 @@ public static enum ConfVars {
HIVE_SERVER2_TEZ_QUEUE_ACCESS_CHECK("hive.server2.tez.queue.access.check",
false,
"Whether to check user access to explicitly specified YARN queues. " +
"yarn.resourcemanager.webapp.address must be configured to use
this."),
+
HIVE_TEZ_QUEUE_METRICS_REFRESH_INTERVAL("hive.tez.queue.metrics.refresh.interval",
"0s",
+ new TimeValidator(TimeUnit.SECONDS),
+ "Interval for refreshing YARN queue resource metrics during Tez query
execution. " +
+ "When set to a positive value (e.g. 10s), displays real-time memory,
vCore, capacity " +
+ "and application metrics for the YARN queue being used. " +
+ "Set to 0 or negative to disable. Minimum effective value is 1
second."),
Review Comment:
The PR description mentions a default refresh interval of 10s in multiple
places, but the actual default here is `0s` (disabled). Please reconcile the PR
description (or update any user-facing docs elsewhere) so the documented
default matches the shipped default.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
