@Override
    public CanEmitBatchOfRecordsChecker getCanEmitBatchOfRecords() {
        return () -> false;
    }
@Test
    void testCanEmitBatchOfRecords() throws Exception {
        AvailabilityProvider.AvailabilityHelper availabilityHelper =
                new AvailabilityProvider.AvailabilityHelper();
        try (StreamTaskMailboxTestHarness<String> testHarness =
                new StreamTaskMailboxTestHarnessBuilder<>(
                                TwoInputStreamTask::new, BasicTypeInfo.STRING_TYPE_INFO)
                        .addInput(BasicTypeInfo.STRING_TYPE_INFO)
                        .addInput(BasicTypeInfo.STRING_TYPE_INFO)
                        .addAdditionalOutput(
                                new ResultPartitionWriterWithAvailabilityHelper(availabilityHelper))
                        .setupOperatorChain(new DuplicatingOperator())
                        .finishForSingletonOperatorChain(IntSerializer.INSTANCE)
                        .build()) {
            StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecordsChecker =
                    testHarness.streamTask.getCanEmitBatchOfRecords();
            testHarness.processAll();

            availabilityHelper.resetAvailable();
            assertThat(canEmitBatchOfRecordsChecker.check()).isFalse();

            // The canEmitBatchOfRecordsChecker should be the false after the record writer is
            // unavailable.
            availabilityHelper.resetUnavailable();
            assertThat(canEmitBatchOfRecordsChecker.check()).isFalse();

            // Restore record writer to available
            availabilityHelper.resetAvailable();
            assertThat(canEmitBatchOfRecordsChecker.check()).isFalse();

            // The canEmitBatchOfRecordsChecker should be the false after add the mail to mail box.
            testHarness.streamTask.mainMailboxExecutor.execute(() -> {}, "mail");
            assertThat(canEmitBatchOfRecordsChecker.check()).isFalse();

            testHarness.processAll();
            assertThat(canEmitBatchOfRecordsChecker.check()).isFalse();
        }
    }