public void inputWatermark(Watermark watermark, int channelIndex, DataOutput<?> output)
            throws Exception {
        final SubpartitionStatus subpartitionStatus;
        if (watermark instanceof InternalWatermark) {
            int subpartitionStatusIndex = ((InternalWatermark) watermark).getSubpartitionIndex();
            subpartitionStatus =
                    subpartitionStatuses.get(channelIndex).get(subpartitionStatusIndex);
        } else {
            subpartitionStatus =
                    subpartitionStatuses.get(channelIndex).get(subpartitionIndexes[channelIndex]);
        }

        // ignore the input watermark if its subpartition, or all subpartitions are idle (i.e.
        // overall the valve is idle).
        if (lastOutputWatermarkStatus.isActive() && subpartitionStatus.watermarkStatus.isActive()) {
            long watermarkMillis = watermark.getTimestamp();

            // if the input watermark's value is less than the last received watermark for its
            // subpartition, ignore it also.
            if (watermarkMillis > subpartitionStatus.watermark) {
                subpartitionStatus.watermark = watermarkMillis;

                if (subpartitionStatus.isWatermarkAligned) {
                    adjustAlignedSubpartitionStatuses(subpartitionStatus);
                } else if (watermarkMillis >= lastOutputWatermark) {
                    // previously unaligned subpartitions are now aligned if its watermark has
                    // caught up
                    markWatermarkAligned(subpartitionStatus);
                }

                // now, attempt to find a new min watermark across all aligned subpartitions
                findAndOutputNewMinWatermarkAcrossAlignedSubpartitions(output);
            }
        }
    }
@Test
    void testMultipleInputYieldsWatermarkOnlyWhenAllChannelsReceivesWatermarks() throws Exception {
        StatusWatermarkOutput valveOutput = new StatusWatermarkOutput();
        StatusWatermarkValve valve = new StatusWatermarkValve(3);

        valve.inputWatermark(new Watermark(0), 0, valveOutput);
        valve.inputWatermark(new Watermark(0), 1, valveOutput);
        assertThat(valveOutput.popLastSeenOutput()).isNull();

        // now, all channels have watermarks
        valve.inputWatermark(new Watermark(0), 2, valveOutput);
        assertThat(valveOutput.popLastSeenOutput()).isEqualTo(new Watermark(0));
        assertThat(valveOutput.popLastSeenOutput()).isNull();
    }