private void initializeInternalStateForSSP(SystemStreamPartition systemStreamPartition) {
    metrics.initMetrics(systemStreamPartition);
    bufferedMessages.putIfAbsent(systemStreamPartition, newBlockingQueue());
    bufferedMessagesSize.putIfAbsent(systemStreamPartition, new AtomicLong(0));
  }
Initializes the metrics and in-memory buffer for the {@param systemStreamPartition}.
@param systemStreamPartition represents the input system stream partition.
samza-api/src/main/java/org/apache/samza/util/BlockingEnvelopeMap.java