abstractdog commented on code in PR #6501: URL: https://github.com/apache/hive/pull/6501#discussion_r3833298888
########## ql/src/test/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/TestQueueMetricsCache.java: ########## @@ -0,0 +1,318 @@ +/* + * 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.yarn.api.records.QueueInfo; +import org.apache.hadoop.yarn.api.records.QueueStatistics; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.when; + +/** + * Test cases for QueueMetricsCache. + */ +@RunWith(MockitoJUnitRunner.class) +public class TestQueueMetricsCache { + + @Mock + private QueueInfo mockQueueInfo; + @Mock + private QueueStatistics mockQueueStats; + + private QueueMetricsCache cache; + + @Before + public void setUp() { + cache = QueueMetricsCache.getInstance(); + cache.shutdown(); // clear any state from previous tests + setupMockQueueInfo(); + } + + @After + public void tearDown() { + cache.shutdown(); + } + + @Test + public void testSingletonInstanceConsistency() { + assertNotNull("Instance should not be null", QueueMetricsCache.getInstance()); + assertSame("getInstance should return same instance", + QueueMetricsCache.getInstance(), QueueMetricsCache.getInstance()); + } + + @Test + public void testGetReturnsNullForNonExistentQueue() { + assertNull("Should return null for non-existent queue", cache.get("test-nonexistent")); + } + + @Test + public void testGetReturnsNullForNullQueueName() { + assertNull("Should return null for null queue name", cache.get(null)); + } + + @Test + public void testPutCreatesNewEntry() { + String queueName = "test-new-entry"; + + // Verify queue doesn't exist yet + assertNull("Queue should not exist initially", cache.get(queueName)); + + // Put creates new entry + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + + QueueMetricsState state = cache.get(queueName); + assertNotNull("Queue state should exist after put", state); + assertNotNull("Snapshot should be available", state.getSnapshot()); + assertEquals("Memory used should match", 1.0f, state.getSnapshot().getMemoryUsedGB(), 0.001f); + } + + @Test + public void testPutUpdatesExistingEntry() { + String queueName = "test-update-entry"; + + // Create initial entry + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + + // Update with new snapshot + when(mockQueueStats.getAllocatedMemoryMB()).thenReturn(4096L); + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 3000L); + + QueueMetricsState state = cache.get(queueName); + assertNotNull("State should exist", state); + assertEquals("Memory should be updated", 4.0f, state.getSnapshot().getMemoryUsedGB(), 0.001f); + } + + @Test + public void testPutWithNullQueueNameIsNoOp() { + // Should not throw exception + cache.put(null, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + // Confirms null queue name was silently ignored - no entry created + assertNull("Null queue name should not create a cache entry", cache.get(null)); + } + + @Test + public void testPutWithNullSnapshotIsNoOp() { + String queueName = "test-null-snapshot-" + System.nanoTime(); + + // Should not throw exception + cache.put(queueName, null, 5000L); + + // Queue should not be created + assertNull("Queue should not exist after put with null snapshot", cache.get(queueName)); + } + + @Test + public void testGetOrCreateCreatesEmptyEntry() { + String queueName = "test-placeholder-" + System.nanoTime(); + + QueueMetricsState state = cache.getOrCreate(queueName, 10000L); + + assertNotNull("Placeholder state should be created", state); + assertNull("Snapshot should be null initially", state.getSnapshot()); + assertEquals("Min interval should match", 10000L, state.getMinRefreshIntervalMs()); + } + + @Test + public void testGetOrCreateWithNullQueueName() { + QueueMetricsState state = cache.getOrCreate(null, 5000L); + + assertNull("Should return null for null queue name", state); + } + + @Test + public void testConcurrentPutPlaceholderRaces() throws Exception { + String queueName = "test-concurrent-placeholder-" + System.nanoTime(); + int threadCount = 10; + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(threadCount); + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + + ConcurrentHashMap<Integer, QueueMetricsState> results = new ConcurrentHashMap<>(); + + // Launch threads that all try to create placeholder simultaneously + for (int i = 0; i < threadCount; i++) { + final int threadId = i; + executor.submit(() -> { + try { + startLatch.await(); // Wait for signal to start + QueueMetricsState state = cache.getOrCreate(queueName, 5000L + threadId * 100); + results.put(threadId, state); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + doneLatch.countDown(); + } + }); + } + + // Start all threads at once + startLatch.countDown(); + + // Wait for completion + assertTrue("Threads should complete", doneLatch.await(5, TimeUnit.SECONDS)); + executor.shutdown(); + + // Verify all threads got the same state instance (putIfAbsent semantics) + QueueMetricsState firstState = results.get(0); + assertNotNull("First state should exist", firstState); + + for (int i = 1; i < threadCount; i++) { + assertSame("All threads should get same state instance", firstState, results.get(i)); + } + + // Verify only one entry exists in cache + QueueMetricsState cachedState = cache.get(queueName); + assertSame("Cached state should match returned state", firstState, cachedState); + } + + @Test + public void testGetActiveQueueCount() { + String queueName1 = "test-count-1"; + String queueName2 = "test-count-2"; + + cache.getOrCreate(queueName1, 5000L); + assertEquals("Count should be 1", 1, cache.getActiveQueueCount()); + + cache.getOrCreate(queueName2, 5000L); + assertEquals("Count should be 2", 2, cache.getActiveQueueCount()); + } + + + @Test + public void testShutdownDoesNotThrow() { + // Should handle gracefully even if called multiple times + cache.shutdown(); + cache.shutdown(); // Second call should also be safe + // After shutdown all entries should be cleared + assertEquals("Cache should be empty after shutdown", 0, cache.getActiveQueueCount()); + } + + @Test + public void testGetOrCreateThenPutUpdatesSnapshot() { + String queueName = "test-placeholder-update-" + System.nanoTime(); + + // Create placeholder first + QueueMetricsState state1 = cache.getOrCreate(queueName, 10000L); + assertNull("Snapshot should be null initially", state1.getSnapshot()); + + // Now put actual snapshot + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + + // Get updated state + QueueMetricsState state2 = cache.get(queueName); + assertSame("Should be same state instance", state1, state2); + assertNotNull("Snapshot should now be populated", state2.getSnapshot()); + assertEquals("Memory should match", 1.0f, state2.getSnapshot().getMemoryUsedGB(), 0.001f); + } + + @Test + public void testConcurrentPutAndGetNoDeadlock() throws Exception { + String queueName = "test-concurrent-ops-" + System.nanoTime(); + int iterationsPerThread = 100; + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(3); + ExecutorService executor = Executors.newFixedThreadPool(3); + AtomicInteger successCount = new AtomicInteger(0); + + // Writer thread 1: getOrCreate + executor.submit(() -> { + try { + startLatch.await(); + for (int i = 0; i < iterationsPerThread; i++) { + cache.getOrCreate(queueName, 5000L); + } + successCount.incrementAndGet(); + } catch (Exception e) { + e.printStackTrace(); + } finally { + doneLatch.countDown(); + } + }); + + // Writer thread 2: put + executor.submit(() -> { + try { + startLatch.await(); + QueueMetricsSnapshot snapshot = new QueueMetricsSnapshot(mockQueueInfo); + for (int i = 0; i < iterationsPerThread; i++) { + cache.put(queueName, snapshot, 3000L); + } + successCount.incrementAndGet(); + } catch (Exception e) { + e.printStackTrace(); Review Comment: nit: LOG instead of `e.printStackTrace()` ########## ql/src/test/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/TestQueueMetricsCache.java: ########## @@ -0,0 +1,318 @@ +/* + * 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.yarn.api.records.QueueInfo; +import org.apache.hadoop.yarn.api.records.QueueStatistics; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.when; + +/** + * Test cases for QueueMetricsCache. + */ +@RunWith(MockitoJUnitRunner.class) +public class TestQueueMetricsCache { + + @Mock + private QueueInfo mockQueueInfo; + @Mock + private QueueStatistics mockQueueStats; + + private QueueMetricsCache cache; + + @Before + public void setUp() { + cache = QueueMetricsCache.getInstance(); + cache.shutdown(); // clear any state from previous tests + setupMockQueueInfo(); + } + + @After + public void tearDown() { + cache.shutdown(); + } + + @Test + public void testSingletonInstanceConsistency() { + assertNotNull("Instance should not be null", QueueMetricsCache.getInstance()); + assertSame("getInstance should return same instance", + QueueMetricsCache.getInstance(), QueueMetricsCache.getInstance()); + } + + @Test + public void testGetReturnsNullForNonExistentQueue() { + assertNull("Should return null for non-existent queue", cache.get("test-nonexistent")); + } + + @Test + public void testGetReturnsNullForNullQueueName() { + assertNull("Should return null for null queue name", cache.get(null)); + } + + @Test + public void testPutCreatesNewEntry() { + String queueName = "test-new-entry"; + + // Verify queue doesn't exist yet + assertNull("Queue should not exist initially", cache.get(queueName)); + + // Put creates new entry + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + + QueueMetricsState state = cache.get(queueName); + assertNotNull("Queue state should exist after put", state); + assertNotNull("Snapshot should be available", state.getSnapshot()); + assertEquals("Memory used should match", 1.0f, state.getSnapshot().getMemoryUsedGB(), 0.001f); + } + + @Test + public void testPutUpdatesExistingEntry() { + String queueName = "test-update-entry"; + + // Create initial entry + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + + // Update with new snapshot + when(mockQueueStats.getAllocatedMemoryMB()).thenReturn(4096L); + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 3000L); + + QueueMetricsState state = cache.get(queueName); + assertNotNull("State should exist", state); + assertEquals("Memory should be updated", 4.0f, state.getSnapshot().getMemoryUsedGB(), 0.001f); + } + + @Test + public void testPutWithNullQueueNameIsNoOp() { + // Should not throw exception + cache.put(null, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + // Confirms null queue name was silently ignored - no entry created + assertNull("Null queue name should not create a cache entry", cache.get(null)); + } + + @Test + public void testPutWithNullSnapshotIsNoOp() { + String queueName = "test-null-snapshot-" + System.nanoTime(); + + // Should not throw exception + cache.put(queueName, null, 5000L); + + // Queue should not be created + assertNull("Queue should not exist after put with null snapshot", cache.get(queueName)); + } + + @Test + public void testGetOrCreateCreatesEmptyEntry() { + String queueName = "test-placeholder-" + System.nanoTime(); + + QueueMetricsState state = cache.getOrCreate(queueName, 10000L); + + assertNotNull("Placeholder state should be created", state); + assertNull("Snapshot should be null initially", state.getSnapshot()); + assertEquals("Min interval should match", 10000L, state.getMinRefreshIntervalMs()); + } + + @Test + public void testGetOrCreateWithNullQueueName() { + QueueMetricsState state = cache.getOrCreate(null, 5000L); + + assertNull("Should return null for null queue name", state); + } + + @Test + public void testConcurrentPutPlaceholderRaces() throws Exception { + String queueName = "test-concurrent-placeholder-" + System.nanoTime(); + int threadCount = 10; + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(threadCount); + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + + ConcurrentHashMap<Integer, QueueMetricsState> results = new ConcurrentHashMap<>(); + + // Launch threads that all try to create placeholder simultaneously + for (int i = 0; i < threadCount; i++) { + final int threadId = i; + executor.submit(() -> { + try { + startLatch.await(); // Wait for signal to start + QueueMetricsState state = cache.getOrCreate(queueName, 5000L + threadId * 100); + results.put(threadId, state); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + doneLatch.countDown(); + } + }); + } + + // Start all threads at once + startLatch.countDown(); + + // Wait for completion + assertTrue("Threads should complete", doneLatch.await(5, TimeUnit.SECONDS)); + executor.shutdown(); + + // Verify all threads got the same state instance (putIfAbsent semantics) + QueueMetricsState firstState = results.get(0); + assertNotNull("First state should exist", firstState); + + for (int i = 1; i < threadCount; i++) { + assertSame("All threads should get same state instance", firstState, results.get(i)); + } + + // Verify only one entry exists in cache + QueueMetricsState cachedState = cache.get(queueName); + assertSame("Cached state should match returned state", firstState, cachedState); + } + + @Test + public void testGetActiveQueueCount() { + String queueName1 = "test-count-1"; + String queueName2 = "test-count-2"; + + cache.getOrCreate(queueName1, 5000L); + assertEquals("Count should be 1", 1, cache.getActiveQueueCount()); + + cache.getOrCreate(queueName2, 5000L); + assertEquals("Count should be 2", 2, cache.getActiveQueueCount()); + } + + + @Test + public void testShutdownDoesNotThrow() { + // Should handle gracefully even if called multiple times + cache.shutdown(); + cache.shutdown(); // Second call should also be safe + // After shutdown all entries should be cleared + assertEquals("Cache should be empty after shutdown", 0, cache.getActiveQueueCount()); + } + + @Test + public void testGetOrCreateThenPutUpdatesSnapshot() { + String queueName = "test-placeholder-update-" + System.nanoTime(); + + // Create placeholder first + QueueMetricsState state1 = cache.getOrCreate(queueName, 10000L); + assertNull("Snapshot should be null initially", state1.getSnapshot()); + + // Now put actual snapshot + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + + // Get updated state + QueueMetricsState state2 = cache.get(queueName); + assertSame("Should be same state instance", state1, state2); + assertNotNull("Snapshot should now be populated", state2.getSnapshot()); + assertEquals("Memory should match", 1.0f, state2.getSnapshot().getMemoryUsedGB(), 0.001f); + } + + @Test + public void testConcurrentPutAndGetNoDeadlock() throws Exception { + String queueName = "test-concurrent-ops-" + System.nanoTime(); + int iterationsPerThread = 100; + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(3); + ExecutorService executor = Executors.newFixedThreadPool(3); + AtomicInteger successCount = new AtomicInteger(0); + + // Writer thread 1: getOrCreate + executor.submit(() -> { + try { + startLatch.await(); + for (int i = 0; i < iterationsPerThread; i++) { + cache.getOrCreate(queueName, 5000L); + } + successCount.incrementAndGet(); + } catch (Exception e) { + e.printStackTrace(); Review Comment: nit: LOG instead of `e.printStackTrace()` ########## common/src/test/org/apache/hadoop/hive/common/log/TestInPlaceUpdate.java: ########## @@ -0,0 +1,193 @@ +/* + * 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.common.log; + +import org.apache.commons.lang3.StringUtils; +import org.junit.Test; + +import java.io.ByteArrayOutputStream; +import java.io.PrintStream; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +/** + * Unit tests for InPlaceUpdate + * <p> + * We capture stdout via a ByteArrayOutputStream and inspect the rendered output. + * These tests verify the rendering layer integration between ProgressMonitor and + * InPlaceUpdate, particularly focusing on separator line positioning when queue + * metrics are displayed. Review Comment: nit: "particularly focusing on separator line positioning when queue metrics are displayed." remove this as it's going to get outdated silently; `TestInPlaceUpdate` makes the reader think it's generic and tests `InPlaceUpdate`, even if it's only testing the queue metrics scenario at the moment ########## ql/src/test/org/apache/hadoop/hive/ql/exec/tez/monitoring/yarnqueue/TestQueueMetricsCache.java: ########## @@ -0,0 +1,318 @@ +/* + * 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.yarn.api.records.QueueInfo; +import org.apache.hadoop.yarn.api.records.QueueStatistics; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.when; + +/** + * Test cases for QueueMetricsCache. + */ +@RunWith(MockitoJUnitRunner.class) +public class TestQueueMetricsCache { + + @Mock + private QueueInfo mockQueueInfo; + @Mock + private QueueStatistics mockQueueStats; + + private QueueMetricsCache cache; + + @Before + public void setUp() { + cache = QueueMetricsCache.getInstance(); + cache.shutdown(); // clear any state from previous tests + setupMockQueueInfo(); + } + + @After + public void tearDown() { + cache.shutdown(); + } + + @Test + public void testSingletonInstanceConsistency() { + assertNotNull("Instance should not be null", QueueMetricsCache.getInstance()); + assertSame("getInstance should return same instance", + QueueMetricsCache.getInstance(), QueueMetricsCache.getInstance()); + } + + @Test + public void testGetReturnsNullForNonExistentQueue() { + assertNull("Should return null for non-existent queue", cache.get("test-nonexistent")); + } + + @Test + public void testGetReturnsNullForNullQueueName() { + assertNull("Should return null for null queue name", cache.get(null)); + } + + @Test + public void testPutCreatesNewEntry() { + String queueName = "test-new-entry"; + + // Verify queue doesn't exist yet + assertNull("Queue should not exist initially", cache.get(queueName)); + + // Put creates new entry + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + + QueueMetricsState state = cache.get(queueName); + assertNotNull("Queue state should exist after put", state); + assertNotNull("Snapshot should be available", state.getSnapshot()); + assertEquals("Memory used should match", 1.0f, state.getSnapshot().getMemoryUsedGB(), 0.001f); + } + + @Test + public void testPutUpdatesExistingEntry() { + String queueName = "test-update-entry"; + + // Create initial entry + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + + // Update with new snapshot + when(mockQueueStats.getAllocatedMemoryMB()).thenReturn(4096L); + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 3000L); + + QueueMetricsState state = cache.get(queueName); + assertNotNull("State should exist", state); + assertEquals("Memory should be updated", 4.0f, state.getSnapshot().getMemoryUsedGB(), 0.001f); + } + + @Test + public void testPutWithNullQueueNameIsNoOp() { + // Should not throw exception + cache.put(null, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + // Confirms null queue name was silently ignored - no entry created + assertNull("Null queue name should not create a cache entry", cache.get(null)); + } + + @Test + public void testPutWithNullSnapshotIsNoOp() { + String queueName = "test-null-snapshot-" + System.nanoTime(); + + // Should not throw exception + cache.put(queueName, null, 5000L); + + // Queue should not be created + assertNull("Queue should not exist after put with null snapshot", cache.get(queueName)); + } + + @Test + public void testGetOrCreateCreatesEmptyEntry() { + String queueName = "test-placeholder-" + System.nanoTime(); + + QueueMetricsState state = cache.getOrCreate(queueName, 10000L); + + assertNotNull("Placeholder state should be created", state); + assertNull("Snapshot should be null initially", state.getSnapshot()); + assertEquals("Min interval should match", 10000L, state.getMinRefreshIntervalMs()); + } + + @Test + public void testGetOrCreateWithNullQueueName() { + QueueMetricsState state = cache.getOrCreate(null, 5000L); + + assertNull("Should return null for null queue name", state); + } + + @Test + public void testConcurrentPutPlaceholderRaces() throws Exception { + String queueName = "test-concurrent-placeholder-" + System.nanoTime(); + int threadCount = 10; + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(threadCount); + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + + ConcurrentHashMap<Integer, QueueMetricsState> results = new ConcurrentHashMap<>(); + + // Launch threads that all try to create placeholder simultaneously + for (int i = 0; i < threadCount; i++) { + final int threadId = i; + executor.submit(() -> { + try { + startLatch.await(); // Wait for signal to start + QueueMetricsState state = cache.getOrCreate(queueName, 5000L + threadId * 100); + results.put(threadId, state); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + doneLatch.countDown(); + } + }); + } + + // Start all threads at once + startLatch.countDown(); + + // Wait for completion + assertTrue("Threads should complete", doneLatch.await(5, TimeUnit.SECONDS)); + executor.shutdown(); + + // Verify all threads got the same state instance (putIfAbsent semantics) + QueueMetricsState firstState = results.get(0); + assertNotNull("First state should exist", firstState); + + for (int i = 1; i < threadCount; i++) { + assertSame("All threads should get same state instance", firstState, results.get(i)); + } + + // Verify only one entry exists in cache + QueueMetricsState cachedState = cache.get(queueName); + assertSame("Cached state should match returned state", firstState, cachedState); + } + + @Test + public void testGetActiveQueueCount() { + String queueName1 = "test-count-1"; + String queueName2 = "test-count-2"; + + cache.getOrCreate(queueName1, 5000L); + assertEquals("Count should be 1", 1, cache.getActiveQueueCount()); + + cache.getOrCreate(queueName2, 5000L); + assertEquals("Count should be 2", 2, cache.getActiveQueueCount()); + } + + + @Test + public void testShutdownDoesNotThrow() { + // Should handle gracefully even if called multiple times + cache.shutdown(); + cache.shutdown(); // Second call should also be safe + // After shutdown all entries should be cleared + assertEquals("Cache should be empty after shutdown", 0, cache.getActiveQueueCount()); + } + + @Test + public void testGetOrCreateThenPutUpdatesSnapshot() { + String queueName = "test-placeholder-update-" + System.nanoTime(); + + // Create placeholder first + QueueMetricsState state1 = cache.getOrCreate(queueName, 10000L); + assertNull("Snapshot should be null initially", state1.getSnapshot()); + + // Now put actual snapshot + cache.put(queueName, new QueueMetricsSnapshot(mockQueueInfo), 5000L); + + // Get updated state + QueueMetricsState state2 = cache.get(queueName); + assertSame("Should be same state instance", state1, state2); + assertNotNull("Snapshot should now be populated", state2.getSnapshot()); + assertEquals("Memory should match", 1.0f, state2.getSnapshot().getMemoryUsedGB(), 0.001f); + } + + @Test + public void testConcurrentPutAndGetNoDeadlock() throws Exception { + String queueName = "test-concurrent-ops-" + System.nanoTime(); + int iterationsPerThread = 100; + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(3); + ExecutorService executor = Executors.newFixedThreadPool(3); + AtomicInteger successCount = new AtomicInteger(0); + + // Writer thread 1: getOrCreate + executor.submit(() -> { + try { + startLatch.await(); + for (int i = 0; i < iterationsPerThread; i++) { + cache.getOrCreate(queueName, 5000L); + } + successCount.incrementAndGet(); + } catch (Exception e) { + e.printStackTrace(); + } finally { + doneLatch.countDown(); + } + }); + + // Writer thread 2: put + executor.submit(() -> { + try { + startLatch.await(); + QueueMetricsSnapshot snapshot = new QueueMetricsSnapshot(mockQueueInfo); + for (int i = 0; i < iterationsPerThread; i++) { + cache.put(queueName, snapshot, 3000L); + } + successCount.incrementAndGet(); + } catch (Exception e) { + e.printStackTrace(); + } finally { + doneLatch.countDown(); + } + }); + + // Reader thread: get + executor.submit(() -> { + try { + startLatch.await(); + for (int i = 0; i < iterationsPerThread; i++) { + cache.get(queueName); + } + successCount.incrementAndGet(); + } catch (Exception e) { + e.printStackTrace(); Review Comment: nit: LOG instead of `e.printStackTrace()` -- 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]
