architjainjain commented on code in PR #6501: URL: https://github.com/apache/hive/pull/6501#discussion_r3718711564
########## ql/src/test/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/TestYarnQueueMetricsCollector.java: ########## @@ -0,0 +1,587 @@ +/* + * 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 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"); + } + Thread.onSpinWait(); // Hint to JVM that this is a spin-wait loop + } + return snapshot; + } + + /** + * Waits for a specific number of invocations with timeout. + */ + private void waitForInvocationCount(Object mock, int minCount, long timeoutMs) { + long startTime = System.currentTimeMillis(); + while (mockingDetails(mock).getInvocations().size() < minCount) { + if (System.currentTimeMillis() - startTime > timeoutMs) { + return; + } + Thread.onSpinWait(); // Hint to JVM that this is a spin-wait loop + } + } + + /** + * Helper method that configures mock objects with standard happy-path values. + */ + private void setupHappyPathMocks() throws Exception { + when(mockQueueStats.getAllocatedMemoryMB()).thenReturn(1024L); + when(mockQueueStats.getAvailableMemoryMB()).thenReturn(1024L); + when(mockQueueStats.getAllocatedVCores()).thenReturn(4L); + when(mockQueueStats.getAvailableVCores()).thenReturn(4L); + when(mockQueueStats.getNumAppsRunning()).thenReturn(1L); + when(mockQueueStats.getNumAppsPending()).thenReturn(0L); + when(mockQueueStats.getAllocatedContainers()).thenReturn(2L); + when(mockQueueStats.getPendingContainers()).thenReturn(0L); + when(mockQueueInfo.getQueueStatistics()).thenReturn(mockQueueStats); + when(mockQueueInfo.getCapacity()).thenReturn(0.5f); + when(mockQueueInfo.getCurrentCapacity()).thenReturn(0.25f); + when(mockYarnClient.getQueueInfo(anyString())).thenReturn(mockQueueInfo); + } + + @Test(expected = IllegalArgumentException.class) + public void testConstructorWithNullYarnClient() { + new YarnQueueMetricsCollector(null, "default", 1000, "query-1", testConf); + } + + @Test(expected = IllegalArgumentException.class) + public void testConstructorWithNullQueueName() { + new YarnQueueMetricsCollector(mockYarnClient, null, 1000, "query-1", testConf); + } + + @Test + public void testSuccessfulMetricsCollection() throws Exception { + setupHappyPathMocks(); + when(mockYarnClient.getQueueInfo("default")).thenReturn(mockQueueInfo); + + YarnQueueMetricsCollector collector = newCollector(mockYarnClient, "default", 10000, "test-query-1"); + try { + QueueMetricsSnapshot snapshot = waitForSnapshot(collector, WAIT_TIMEOUT_MS); + + assertNotNull("Snapshot should not be null", snapshot); + assertEquals("Memory used should be 1GB", 1.0f, snapshot.getMemoryUsedGB(), 0.1f); + assertEquals("Memory total should be 2GB (1+1)", 2.0f, snapshot.getMemoryTotalGB(), 0.1f); + assertEquals("VCores used should be 4", 4, snapshot.getVCoresUsed()); + assertEquals("VCores total should be 8 (4+4)", 8, snapshot.getVCoresTotal()); + assertEquals("Running apps should be 1", 1, snapshot.getRunningApps()); + assertEquals("Pending apps should be 0", 0, snapshot.getPendingApps()); + assertEquals("Allocated containers should be 2", 2, snapshot.getAllocatedContainers()); + assertEquals("Pending containers should be 0", 0, snapshot.getPendingContainers()); + assertEquals("Capacity should be 50%", 50.0f, snapshot.getCapacityPercentage(), 0.1f); + assertEquals("Current capacity should be 25%", 25.0f, snapshot.getCurrentCapacityPercentage(), 0.1f); + assertEquals("Memory percentage", "50.00%", snapshot.getMemoryPercentage()); + assertEquals("VCores percentage", "50.00%", snapshot.getVCoresPercentage()); + } finally { + collector.shutdown(); + } + } + + @Test + public void testMetricsCollectionWithNullQueueInfo() throws Exception { + when(mockYarnClient.getQueueInfo("nonexistent")).thenReturn(null); + + YarnQueueMetricsCollector collector = newCollector(mockYarnClient, "nonexistent", 10000, "test-query-2"); + try { + assertNull("Snapshot should be null for nonexistent queue", collector.getLatestSnapshot()); + } finally { + collector.shutdown(); + } + } + + @Test + public void testMetricsCollectionWithNullQueueStatistics() throws Exception { + when(mockQueueInfo.getQueueStatistics()).thenReturn(null); + when(mockQueueInfo.getCapacity()).thenReturn(0.5f); + when(mockQueueInfo.getCurrentCapacity()).thenReturn(0.0f); + when(mockYarnClient.getQueueInfo("default")).thenReturn(mockQueueInfo); + + YarnQueueMetricsCollector collector = newCollector(mockYarnClient, "default", 10000, "test-query-3"); + try { + QueueMetricsSnapshot snapshot = waitForSnapshot(collector, WAIT_TIMEOUT_MS); + assertNotNull("Snapshot should not be null", snapshot); + assertEquals("Memory used should be 0", 0.0f, snapshot.getMemoryUsedGB(), 0.01f); + assertEquals("Memory total should be 0", 0.0f, snapshot.getMemoryTotalGB(), 0.01f); + assertEquals("VCores used should be 0", 0, snapshot.getVCoresUsed()); + assertEquals("VCores total should be 0", 0, snapshot.getVCoresTotal()); + assertEquals("Capacity should still be 50%", 50.0f, snapshot.getCapacityPercentage(), 0.1f); + assertEquals("Current capacity should be 0%", 0.0f, snapshot.getCurrentCapacityPercentage(), 0.1f); + } finally { + collector.shutdown(); + } + } + + @Test + public void testPercentageCalculationWithZeroTotal() { + // Setup with zero totals + when(mockQueueStats.getAllocatedMemoryMB()).thenReturn(0L); + when(mockQueueStats.getAvailableMemoryMB()).thenReturn(0L); + when(mockQueueStats.getAllocatedVCores()).thenReturn(0L); + when(mockQueueStats.getAvailableVCores()).thenReturn(0L); + when(mockQueueStats.getNumAppsRunning()).thenReturn(0L); + when(mockQueueStats.getNumAppsPending()).thenReturn(0L); + when(mockQueueStats.getAllocatedContainers()).thenReturn(0L); + when(mockQueueStats.getPendingContainers()).thenReturn(0L); + when(mockQueueInfo.getQueueStatistics()).thenReturn(mockQueueStats); + when(mockQueueInfo.getCapacity()).thenReturn(0.0f); + when(mockQueueInfo.getCurrentCapacity()).thenReturn(0.0f); + + QueueMetricsSnapshot snapshot = + new QueueMetricsSnapshot(mockQueueInfo); + + // Should return "N/A" for percentages when total is zero + assertEquals("Memory percentage should be N/A", "N/A", snapshot.getMemoryPercentage()); + assertEquals("VCores percentage should be N/A", "N/A", snapshot.getVCoresPercentage()); + } + + @Test + public void testShutdownIdempotency() throws Exception { + when(mockYarnClient.getQueueInfo("default")).thenReturn(mockQueueInfo); + + YarnQueueMetricsCollector collector = newCollector(mockYarnClient, "default", 10000, "test-query-4"); + collector.shutdown(); + collector.shutdown(); // second call must be safe + assertTrue("Multiple shutdowns should be safe", true); + } + + @Test + public void testExceptionDuringCollection() throws Exception { + when(mockYarnClient.getQueueInfo("default")) + .thenThrow(new RuntimeException("RM unavailable")); + + YarnQueueMetricsCollector collector = newCollector(mockYarnClient, "default", 10000, "test-query-5"); + try { + assertNull("Snapshot should be null after exception", collector.getLatestSnapshot()); + } finally { + collector.shutdown(); + } + } + + @Test + public void testQueueNameRetrieval() throws Exception { + when(mockYarnClient.getQueueInfo(anyString())).thenReturn(mockQueueInfo); + when(mockQueueInfo.getQueueStatistics()).thenReturn(null); + when(mockQueueInfo.getCapacity()).thenReturn(0.5f); + + YarnQueueMetricsCollector collector = newCollector(mockYarnClient, "production", 10000, "test-query-6"); + try { + assertEquals("Queue name should match", "production", collector.getQueueName()); + } finally { + collector.shutdown(); + } + } + + @Test + public void testMemoryAndVCoreCalculation() { + // Test with specific values to verify calculation + when(mockQueueStats.getAllocatedMemoryMB()).thenReturn(5120L); // 5GB used + when(mockQueueStats.getAvailableMemoryMB()).thenReturn(15360L); // 15GB available + when(mockQueueStats.getAllocatedVCores()).thenReturn(50L); + when(mockQueueStats.getAvailableVCores()).thenReturn(150L); + when(mockQueueStats.getNumAppsRunning()).thenReturn(3L); + when(mockQueueStats.getNumAppsPending()).thenReturn(2L); + when(mockQueueStats.getAllocatedContainers()).thenReturn(10L); + when(mockQueueStats.getPendingContainers()).thenReturn(7L); + when(mockQueueInfo.getQueueStatistics()).thenReturn(mockQueueStats); + when(mockQueueInfo.getCapacity()).thenReturn(0.2f); // 20% + when(mockQueueInfo.getCurrentCapacity()).thenReturn(0.05f); // 5% + + QueueMetricsSnapshot snapshot = + new QueueMetricsSnapshot(mockQueueInfo); + + // Total = Used + Available + assertEquals("Memory used", 5.0f, snapshot.getMemoryUsedGB(), 0.01f); + assertEquals("Memory total", 20.0f, snapshot.getMemoryTotalGB(), 0.01f); // 5+15 + assertEquals("Memory percentage", "25.00%", snapshot.getMemoryPercentage()); // 5/20 + + assertEquals("VCores used", 50, snapshot.getVCoresUsed()); + assertEquals("VCores total", 200, snapshot.getVCoresTotal()); // 50+150 + assertEquals("VCores percentage", "25.00%", snapshot.getVCoresPercentage()); // 50/200 + + assertEquals("Running apps", 3, snapshot.getRunningApps()); + assertEquals("Pending apps", 2, snapshot.getPendingApps()); + assertEquals("Allocated containers", 10, snapshot.getAllocatedContainers()); + assertEquals("Pending containers", 7, snapshot.getPendingContainers()); + assertEquals("Capacity", 20.0f, snapshot.getCapacityPercentage(), 0.01f); + assertEquals("Current capacity", 5.0f, snapshot.getCurrentCapacityPercentage(), 0.01f); + } + + @Test(expected = IllegalArgumentException.class) + public void testQueueMetricsSnapshotWithNullQueueInfo() { + new QueueMetricsSnapshot(null); + } + + // ------------------------------------------------------------------------- + // Tests for Issue #1: Jitter on initial delay (Thundering Herd prevention) + // ------------------------------------------------------------------------- + // Note: Jitter is implicitly tested by all tests that successfully create collectors. + // Explicit jitter distribution testing would require reflection to access private + // scheduling details, which is fragile and not worth the maintenance cost. Review Comment: done -- 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]
