void update(WatermarkMessage watermarkMessage, SystemStreamPartition ssp) {
    WatermarkState state = watermarkStates.get(ssp);
    if (state != null) {
      state.update(watermarkMessage.getTimestamp(), watermarkMessage.getTaskName());
    } else {
      LOG.error("SSP {} doesn't have watermark states", ssp);
    }
  }
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