@Disabled // Manual running only, too slow for CI
    @Test
    void testFlushAndAggregatePerformance() throws SQLException, InterruptedException {

        final LogExecutionTime totalRunTime = LogExecutionTime.start();
        final List<String> tagNames = List.of("Tag1", "Tag2");
        // Make sure each flush call executes two and a bit batches
        final int flushLimit = (int) (sqlStatisticsConfig.getStatisticFlushBatchSize() * 2.5);
        final int iterations = 100_000;
        final int keysPerIteration = 2;
        final int tagValuesPerKey = 2;
        final long initialEventDeltaMs = 1_000;
        final long eventDeltaDeltaMs = 20;
        final int flushWorkerThreads = 4;
        final int flushWorkers = 4;
        final int expectedEventCount = iterations * keysPerIteration * tagValuesPerKey * flushWorkers;
        final Instant startTime = Instant.from(
                ZonedDateTime.of(
                        2022,
                        6,
                        30,
                        12,
                        0,
                        0,
                        0,
                        ZoneOffset.UTC));
        final LongAdder totalEventCount = new LongAdder();
        sqlStatisticAggregationManager.setStage1BatchSize(500_000);
        sqlStatisticAggregationManager.setStage2BatchSize(500_000);

        LOGGER.info("Start time: {}", startTime);

        clearAllStatTables();

        assertThat(getStatValSrcRowCount())
                .isEqualTo(0);

        final SQLStatisticFlushTaskHandler sqlStatisticFlushTaskHandler = new SQLStatisticFlushTaskHandler(
                sqlStatisticValueBatchSaveService, taskContextFactory, securityContext, () -> sqlStatisticsConfig);
        final CountDownLatch doneFirstFlush = new CountDownLatch(1);

        final Runnable flushRunnable = () -> {
            final AtomicReference<SQLStatisticAggregateMap> aggregateMapRef = new AtomicReference<>(
                    new SQLStatisticAggregateMap());

            final Runnable doFlush = () -> {
                final SQLStatisticAggregateMap map = aggregateMapRef.get();
                LOGGER.logDurationIfInfoEnabled(() ->
                                sqlStatisticFlushTaskHandler.exec(map),
                        "Flush");
                totalEventCount.add(map.size());
                aggregateMapRef.set(new SQLStatisticAggregateMap());
            };
            Instant time = startTime;
            long eventDeltaMs = initialEventDeltaMs;
            long eventCount = 0;
            for (int i = 0; i < iterations; i++) {
//                time = startTime.minus(i * eventFreqSecs, ChronoUnit.SECONDS);
                time = time.minus(eventDeltaMs, ChronoUnit.MILLIS);
                // Make the gaps between events a bit bigger each time so we span a longer period
                eventDeltaMs += eventDeltaDeltaMs;
                for (int j = 0; j < keysPerIteration; j++) {
                    for (int k = 0; k < tagValuesPerKey; k++) {

                        final int finalK = k;
                        final List<StatisticTag> tags = tagNames.stream()
                                .map(name -> new StatisticTag(name, "Val" + finalK))
                                .collect(Collectors.toList());

                        final StatisticEvent event = StatisticEvent.createCount(
                                time.toEpochMilli(), "StatKey" + j, tags, 10);

                        try {
                            aggregateMapRef.get().addRolledUpEvent(
                                    new RolledUpStatisticEvent(event),
                                    1_000);
                        } catch (StatisticsEventValidationException e) {
                            throw new RuntimeException(e);
                        }
                        eventCount++;
                        if (eventCount >= flushLimit) {
                            doFlush.run();
                            doneFirstFlush.countDown();
                            eventCount = 0;
                        }
                    }
                }
            }

            if (aggregateMapRef.get().size() > 0) {
                doFlush.run();
                doneFirstFlush.countDown();
            }
            LOGGER.info("Earliest time: {}, eventDeltaMs: {}", time, Duration.ofMillis(eventDeltaMs));
        };

        final Executor flushWorkersExecutor = Executors.newFixedThreadPool(flushWorkerThreads);
        final CountDownLatch countDownLatch = new CountDownLatch(flushWorkers);

        for (int i = 0; i < flushWorkers; i++) {
            final int finalI = i;
            CompletableFuture.runAsync(() -> {
                LOGGER.info("Running flush worker {}", finalI);
                flushRunnable.run();
                countDownLatch.countDown();
                if (countDownLatch.getCount() == 0) {
                    LOGGER.info("Finished last flush");
                }
            }, flushWorkersExecutor);
        }

        // Keep running aggregation until all flushes have finished
        final LogExecutionTime totalAggTime = LogExecutionTime.start();
        Instant aggregationNow = startTime;

        // Don't start aggregating till
        doneFirstFlush.await();
        ThreadUtil.sleep(100);
        while (countDownLatch.getCount() > 0) {
            sqlStatisticAggregationManager.aggregate(aggregationNow);
        }
        LOGGER.info("Flushing finished, running final aggregation");
        sqlStatisticAggregationManager.aggregate(aggregationNow);

        LOGGER.info("Running aggregation 2 hours after start time");
        aggregationNow = aggregationNow.plus(2, ChronoUnit.HOURS);
        // This ensures we run stage 2 agg.
        sqlStatisticAggregationManager.aggregate(aggregationNow);

        LOGGER.info("Running aggregation 2 days after the last time");
        aggregationNow = aggregationNow.plus(2, ChronoUnit.DAYS);
        sqlStatisticAggregationManager.aggregate(aggregationNow);

        LOGGER.info("Running aggregation 40 days after the last time");
        aggregationNow = aggregationNow.plus(40, ChronoUnit.DAYS);
        sqlStatisticAggregationManager.aggregate(aggregationNow);

        LOGGER.info("------------------------------------------------------------");
        LOGGER.info("""
                        Test settings:
                        iterations: {},
                        keysPerIteration: {},
                        tagValuesPerKey: {},
                        initialEventDeltaMs: {},
                        eventDeltaDeltaMs: {},
                        flushWorkerThreads: {},
                        flushWorkers: {}""",
                iterations,
                keysPerIteration,
                tagValuesPerKey,
                initialEventDeltaMs,
                eventDeltaDeltaMs,
                flushWorkerThreads,
                flushWorkers);
        LOGGER.info("Total event count flushed: {}", totalEventCount);
        LOGGER.info("Total agg time: {}", totalAggTime.getDuration());
        LOGGER.info("Total test run time: {}", totalRunTime.getDuration());
        LOGGER.info("------------------------------------------------------------");

        assertThat(getStatValSrcRowCount())
                .isEqualTo(0);
        assertThat(totalEventCount.longValue())
                .isEqualTo(expectedEventCount);
    }
The aim of this test is for multiple threads to be flushing stats into SQL_STAT_VAL_SRC
and for the main thread to be constantly running the aggregation to exercise contention
on any db locks.
Useful for testing the performance of the two process when running concurrently.
stroom-app/src/test/java/stroom/statistics/impl/sql/TestSQLStatisticFlushTaskHandler.java