void updateAggregateMetric(SystemStreamPartition ssp, long time) {
    if (intermediateSsps.contains(ssp)) {
      // Only report the aggregates watermarks for intermediate streams
      // to reduce the amount of metrics
      watermarkMetrics.setAggregateTime(ssp, time);

      final WatermarkState state = watermarkStates.get(ssp);
      if (state != null && state.getQuorumCount() != 0) {
        watermarkMetrics.setQuorumCount(ssp.getSystemStream(), state.getQuorumCount());
      }
    }
  }
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