@Override
    public final Environment getEnvironment() {
        return environment;
    }
@Test
    void testMailboxMetricsMeasurement() throws Exception {
        final int numMails = 10, sleepTime = 5;
        StreamTaskMailboxTestHarnessBuilder<Integer> builder =
                new StreamTaskMailboxTestHarnessBuilder<>(
                                OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
                        .addInput(BasicTypeInfo.INT_TYPE_INFO)
                        .setupOutputForSingletonOperatorChain(
                                new TestBoundedOneInputStreamOperator());
        try (StreamTaskMailboxTestHarness<Integer> harness = builder.build()) {
            Histogram mailboxLatencyMetric =
                    harness.streamTask
                            .getEnvironment()
                            .getMetricGroup()
                            .getIOMetricGroup()
                            .getMailboxLatency();
            Gauge<Integer> mailboxSizeMetric =
                    harness.streamTask
                            .getEnvironment()
                            .getMetricGroup()
                            .getIOMetricGroup()
                            .getMailboxSize();
            long startTime = SystemClock.getInstance().relativeTimeMillis();
            harness.streamTask.mailboxProcessor.getMailboxMetricsControl().measureMailboxLatency();
            for (int i = 0; i < numMails; ++i) {
                harness.streamTask.mainMailboxExecutor.execute(
                        () -> Thread.sleep(sleepTime), "add value");
            }
            harness.streamTask.mailboxProcessor.getMailboxMetricsControl().measureMailboxLatency();

            assertThat(mailboxSizeMetric.getValue()).isGreaterThanOrEqualTo(numMails);
            assertThat(mailboxLatencyMetric.getCount()).isZero();

            harness.processAll();
            long endTime = SystemClock.getInstance().relativeTimeMillis();

            assertThat(mailboxSizeMetric.getValue()).isZero();
            assertThat(mailboxLatencyMetric.getCount()).isEqualTo(2L);
            assertThat(mailboxLatencyMetric.getStatistics().getMax())
                    .isBetween((long) (sleepTime * numMails), endTime - startTime);
        }
    }