public TaskIOMetricGroup(TaskMetricGroup parent) {
        this(parent, SystemClock.getInstance());
    }
@Test
    void testTaskIOMetricGroup() throws InterruptedException {
        TaskMetricGroup task = UnregisteredMetricGroups.createUnregisteredTaskMetricGroup();
        ManualClock clock = new ManualClock(System.currentTimeMillis());
        TaskIOMetricGroup taskIO = new TaskIOMetricGroup(task, clock);

        // test initializing time
        final long initializationTime = 100L;
        assertThat(taskIO.getTaskInitializationDuration()).isEqualTo(0L);
        taskIO.markTaskInitializationStarted();
        clock.advanceTime(Duration.ofMillis(initializationTime));
        assertThat(taskIO.getTaskInitializationDuration()).isGreaterThan(0L);
        long initializationDuration = taskIO.getTaskInitializationDuration();
        taskIO.markTaskStart();
        assertThat(taskIO.getTaskInitializationDuration()).isEqualTo(initializationDuration);

        taskIO.setEnableBusyTime(true);
        taskIO.markTaskStart();
        final long startTime = clock.absoluteTimeMillis();

        // test initializing time remains unchanged after running
        final long runningTime = 200L;
        clock.advanceTime(Duration.ofMillis(runningTime));
        assertThat(taskIO.getTaskInitializationDuration()).isEqualTo(initializationDuration);

        // test counter forwarding
        assertThat(taskIO.getNumRecordsInCounter()).isNotNull();
        assertThat(taskIO.getNumRecordsOutCounter()).isNotNull();

        Counter c1 = new SimpleCounter();
        c1.inc(32L);
        Counter c2 = new SimpleCounter();
        c2.inc(64L);

        taskIO.reuseRecordsInputCounter(c1);
        taskIO.reuseRecordsOutputCounter(c2);

        assertThat(taskIO.getNumRecordsInCounter().getCount()).isEqualTo(32L);
        assertThat(taskIO.getNumRecordsOutCounter().getCount()).isEqualTo(64L);

        // test IOMetrics instantiation
        taskIO.getNumBytesInCounter().inc(100L);
        taskIO.getNumBytesOutCounter().inc(250L);
        taskIO.getNumBuffersOutCounter().inc(3L);
        taskIO.getIdleTimeMsPerSecond().markStart();
        taskIO.getSoftBackPressuredTimePerSecond().markStart();
        long softSleepTime = 2L;
        clock.advanceTime(Duration.ofMillis(softSleepTime));
        taskIO.getIdleTimeMsPerSecond().markEnd();
        taskIO.getSoftBackPressuredTimePerSecond().markEnd();

        long hardSleepTime = 4L;
        taskIO.getHardBackPressuredTimePerSecond().markStart();
        clock.advanceTime(Duration.ofMillis(hardSleepTime));
        taskIO.getHardBackPressuredTimePerSecond().markEnd();

        long ioSleepTime = 3L;
        taskIO.getChangelogBusyTimeMsPerSecond().markStart();
        clock.advanceTime(Duration.ofMillis(ioSleepTime));
        taskIO.getChangelogBusyTimeMsPerSecond().markEnd();

        IOMetrics io = taskIO.createSnapshot();
        assertThat(io.getNumRecordsIn()).isEqualTo(32L);
        assertThat(io.getNumRecordsOut()).isEqualTo(64L);
        assertThat(io.getNumBytesIn()).isEqualTo(100L);
        assertThat(io.getNumBytesOut()).isEqualTo(250L);
        assertThat(taskIO.getNumBuffersOutCounter().getCount()).isEqualTo(3L);
        assertThat(taskIO.getIdleTimeMsPerSecond().getAccumulatedCount())
                .isEqualTo(io.getAccumulateIdleTime());
        assertThat(
                        taskIO.getHardBackPressuredTimePerSecond().getAccumulatedCount()
                                + taskIO.getSoftBackPressuredTimePerSecond().getAccumulatedCount())
                .isEqualTo(io.getAccumulateBackPressuredTime());
        assertThat(io.getAccumulateBusyTime())
                .isEqualTo(
                        clock.absoluteTimeMillis()
                                - startTime
                                - io.getAccumulateIdleTime()
                                - io.getAccumulateBackPressuredTime());
        assertThat(taskIO.getIdleTimeMsPerSecond().getCount())
                .isGreaterThanOrEqualTo(softSleepTime);
        assertThat(taskIO.getSoftBackPressuredTimePerSecond().getCount())
                .isGreaterThanOrEqualTo(softSleepTime);
        assertThat(taskIO.getHardBackPressuredTimePerSecond().getCount())
                .isGreaterThanOrEqualTo(hardSleepTime);
        assertThat(taskIO.getChangelogBusyTimeMsPerSecond().getCount())
                .isGreaterThanOrEqualTo(ioSleepTime);
    }