public void inputWatermarkStatus(
            WatermarkStatus watermarkStatus, int channelIndex, DataOutput<?> output)
            throws Exception {
        // Shared input channel is only enabled in batch jobs, which do not have watermark status
        // events.
        Preconditions.checkState(!isInputChannelShared);
        SubpartitionStatus subpartitionStatus =
                subpartitionStatuses.get(channelIndex).get(subpartitionIndexes[channelIndex]);

        // It is supposed that WatermarkStatus will not appear in jobs where one input channel
        // consumes multiple subpartitions, so we do not need to map channelIndex into
        // subpartitionStatusIndex for now, like what is done on Watermarks.

        // only account for watermark status inputs that will result in a status change for the
        // subpartition
        if (watermarkStatus.isIdle() && subpartitionStatus.watermarkStatus.isActive()) {
            // handle active -> idle toggle for the subpartition
            subpartitionStatus.watermarkStatus = WatermarkStatus.IDLE;

            // the subpartition is now idle, therefore not aligned
            markWatermarkUnaligned(subpartitionStatus);

            // if all subpartitions of the valve are now idle, we need to output an idle stream
            // status from the valve (this also marks the valve as idle)
            if (!SubpartitionStatus.hasActiveSubpartitions(subpartitionStatuses)) {

                // now that all subpartitions are idle and no subpartitions will continue to advance
                // its
                // watermark,
                // we should "flush" all watermarks across all subpartitions; effectively, this
                // means
                // emitting
                // the max watermark across all subpartitions as the new watermark. Also, since we
                // already try to advance
                // the min watermark as subpartitions individually become IDLE, here we only need to
                // perform the flush
                // if the watermark of the last active subpartition that just became idle is the
                // current
                // min watermark.
                if (subpartitionStatus.watermark == lastOutputWatermark) {
                    findAndOutputMaxWatermarkAcrossAllSubpartitions(output);
                }

                lastOutputWatermarkStatus = WatermarkStatus.IDLE;
                output.emitWatermarkStatus(lastOutputWatermarkStatus);
            } else if (subpartitionStatus.watermark == lastOutputWatermark) {
                // if the watermark of the subpartition that just became idle equals the last output
                // watermark (the previous overall min watermark), we may be able to find a new
                // min watermark from the remaining aligned subpartitions
                findAndOutputNewMinWatermarkAcrossAlignedSubpartitions(output);
            }
        } else if (watermarkStatus.isActive() && subpartitionStatus.watermarkStatus.isIdle()) {
            // handle idle -> active toggle for the subpartition
            subpartitionStatus.watermarkStatus = WatermarkStatus.ACTIVE;

            // if the last watermark of the subpartition, before it was marked idle, is still
            // larger than
            // the overall last output watermark of the valve, then we can set the subpartition to
            // be
            // aligned already.
            if (subpartitionStatus.watermark >= lastOutputWatermark) {
                markWatermarkAligned(subpartitionStatus);
            }

            // if the valve was previously marked to be idle, mark it as active and output an active
            // stream
            // status because at least one of the subpartitions is now active
            if (lastOutputWatermarkStatus.isIdle()) {
                lastOutputWatermarkStatus = WatermarkStatus.ACTIVE;
                output.emitWatermarkStatus(lastOutputWatermarkStatus);
            }
        }
    }
@Test
    void testSingleInputWatermarkStatusToggling() throws Exception {
        StatusWatermarkOutput valveOutput = new StatusWatermarkOutput();
        StatusWatermarkValve valve = new StatusWatermarkValve(1);

        valve.inputWatermarkStatus(WatermarkStatus.ACTIVE, 0, valveOutput);
        // this also implicitly verifies that input channels start as ACTIVE
        assertThat(valveOutput.popLastSeenOutput()).isNull();

        valve.inputWatermarkStatus(WatermarkStatus.IDLE, 0, valveOutput);
        assertThat(valveOutput.popLastSeenOutput()).isEqualTo(WatermarkStatus.IDLE);

        valve.inputWatermarkStatus(WatermarkStatus.IDLE, 0, valveOutput);
        assertThat(valveOutput.popLastSeenOutput()).isNull();

        valve.inputWatermarkStatus(WatermarkStatus.ACTIVE, 0, valveOutput);
        assertThat(valveOutput.popLastSeenOutput()).isEqualTo(WatermarkStatus.ACTIVE);
        assertThat(valveOutput.popLastSeenOutput()).isNull();
    }