testMetricsProvideCorrectUtilizationAndActiveExecutorCount() — netty Function Reference

Source Code

common/src/test/java/io/netty/util/concurrent/AutoScalingEventExecutorChooserFactoryTest.java lines 273–339

    @Test
    @Timeout(30)
    void testMetricsProvideCorrectUtilizationAndActiveExecutorCount() throws InterruptedException {
        TestEventExecutorGroup group = new TestEventExecutorGroup(1, 3, 50, TimeUnit.MILLISECONDS);
        try {
            startAllExecutors(group);
            long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
            while (group.activeExecutorCount() > 1 && System.nanoTime() < deadline) {
                Thread.sleep(50);
            }
            assertEquals(1, group.activeExecutorCount(), "Should have scaled down to 1 active executor.");

            TestEventExecutor activeExecutor = null;
            for (EventExecutor exec : group) {
                if (!exec.isSuspended()) {
                    activeExecutor = (TestEventExecutor) exec;
                    break;
                }
            }
            if (activeExecutor == null) {
                fail("Could not find an active executor.");
            }

            activeExecutor.setHighLoad(true);

            while (System.nanoTime() < deadline) {
                List<AutoScalingUtilizationMetric> utilizationMetrics = group.executorUtilizations();
                TestEventExecutor finalActiveExecutor = activeExecutor;
                double utilization = utilizationMetrics.stream()
                                                       .filter(metric -> metric.executor().equals(finalActiveExecutor))
                                                       .findFirst()
                        .map(AutoScalingUtilizationMetric::utilization)
                        .orElse(0.0);
                if (utilization > 0.4) {
                    break;
                }
                Thread.sleep(50);
            }

            assertEquals(1, group.activeExecutorCount(), "Active count should still be 1 before scaling up.");

            List<AutoScalingUtilizationMetric> utilizationMetrics = group.executorUtilizations();
            assertEquals(3, utilizationMetrics.size(), "Utilization list should report on all executors.");

            TestEventExecutor finalActiveExecutor2 = activeExecutor;
            double activeUtilization = utilizationMetrics.stream()
                                                         .filter(metric -> metric.executor()
                                                                                 .equals(finalActiveExecutor2))
                                                         .findFirst()
                                                         .map(AutoScalingUtilizationMetric::utilization)
                                                         .orElse(0.0);
            assertTrue(activeUtilization > 0.4,
                       "Active executor should have utilization above the scale-down threshold. " +
                       "Was: " + activeUtilization);

            TestEventExecutor finalActiveExecutor1 = activeExecutor;
            utilizationMetrics.stream()
                              .filter(metric -> metric.executor() != finalActiveExecutor1)
                              .forEach(metric -> {
                                  assertTrue(metric.executor().isSuspended(), "Other executors should be suspended.");
                                  assertEquals(0.0, metric.utilization(),
                                               "Suspended executor should have 0.0 utilization.");
                              });
        } finally {
            group.shutdownGracefully().syncUninterruptibly();
        }
    }