code: public void inputWatermark(Watermark watermark, int channelIndex) {
  		// ignore the input watermark if its input channel, or all input channels are idle (i.e. overall the valve is idle).
  		if (lastOutputStreamStatus.isActive() && channelStatuses[channelIndex].streamStatus.isActive()) {
  			long watermarkMillis = watermark.getTimestamp();
  
  			// if the input watermark's value is less than the last received watermark for its input channel, ignore it also.
  			if (watermarkMillis > channelStatuses[channelIndex].watermark) {
  				channelStatuses[channelIndex].watermark = watermarkMillis;
  
  				// previously unaligned input channels are now aligned if its watermark has caught up
  				if (!channelStatuses[channelIndex].isWatermarkAligned && watermarkMillis >= lastOutputWatermark) {
  					channelStatuses[channelIndex].isWatermarkAligned = true;
  				}
  
  				// now, attempt to find a new min watermark across all aligned channels
  				findAndOutputNewMinWatermarkAcrossAlignedChannels();
  			}
  		}
  	}
docs: Feed a {@link Watermark} into the valve. If the input triggers the valve to output a new Watermark,
  {@link ValveOutputHandler#handleWatermark(Watermark)} will be called to process the new Watermark.
  
  @param watermark the watermark to feed to the valve
  @param channelIndex the index of the channel that the fed watermark belongs to (index starting from 0)