long getWatermarkPerSSP(SystemStreamPartition ssp) {
    return watermarkStates.get(ssp).getWatermarkTime();
  }
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