public CompletableFuture<Void> stop(StopMode mode) {
        switch (operatingMode) {
            case WAITING_FOR_ALIGNMENT:
            case OUTPUT_NOT_INITIALIZED:
            case READING:
                this.operatingMode =
                        mode == StopMode.DRAIN
                                ? OperatingMode.SOURCE_DRAINED
                                : OperatingMode.SOURCE_STOPPED;
                availabilityHelper.forceStop();
                if (this.operatingMode == OperatingMode.SOURCE_STOPPED) {
                    stopInternalServices();
                    finished.complete(null);
                    return finished;
                }
                break;
        }
        return finished;
    }
@Test
    void testStop() throws Exception {
        // Initialize the operator.
        operator.initializeState(context.createStateContext());
        // Open the operator.
        operator.open();
        // The source reader should have been assigned a split.
        assertThat(mockSourceReader.getAssignedSplits())
                .containsExactly(SourceOperatorTestContext.MOCK_SPLIT);

        CollectingDataOutput<Integer> dataOutput = new CollectingDataOutput<>();
        assertThat(operator.emitNext(dataOutput)).isEqualTo(DataInputStatus.NOTHING_AVAILABLE);
        assertThat(operator.isAvailable()).isFalse();

        CompletableFuture<Void> sourceStopped = operator.stop(StopMode.DRAIN);
        assertThat(operator.isAvailable()).isTrue();
        assertThat(sourceStopped).isNotDone();
        assertThat(operator.emitNext(dataOutput)).isEqualTo(DataInputStatus.END_OF_DATA);
        operator.finish();
        assertThat(sourceStopped).isDone();
    }