long getWatermark(SystemStream systemStream) {
    return watermarkStates.entrySet().stream()
        .filter(entry -> entry.getKey().getSystemStream().equals(systemStream))
        .map(entry -> entry.getValue().getWatermarkTime())
        .min(Long::compare)
        .orElse(WATERMARK_NOT_EXIST);
  }
Update the state upon receiving a watermark message.
@param watermarkMessage message of {@link WatermarkMessage}
@param ssp system stream partition
@return true iff the stream has a new watermark
samza-core/src/main/java/org/apache/samza/operators/impl/WatermarkStates.java