private IncomingMessageEnvelope fetchEnvelope() {
      PendingEnvelope pendingEnvelope = pendingEnvelopeQueue.remove();
      int queueSize = pendingEnvelopeQueue.size();
      taskMetrics.pendingMessages().set(queueSize);
      log.trace("fetch envelope ssp {} offset {} to process.",
          pendingEnvelope.envelope.getSystemStreamPartition(elasticityFactor), pendingEnvelope.envelope.getOffset());
      log.debug("Task {} pending envelopes count is {} after fetching.", taskName, queueSize);

      if (pendingEnvelope.markProcessed()) {
        SystemStreamPartition partition = pendingEnvelope.envelope.getSystemStreamPartition(elasticityFactor);
        consumerMultiplexer.tryUpdate(partition);
        log.debug("Update chooser for {}", partition);
      }
      return pendingEnvelope.envelope;
    }
Fetch the pending envelope in the pending queue for the task to process.
Update the chooser for flow control on the SSP level. Once it's updated, the RunLoop
will be able to choose new messages from this SSP for the task to process. Note that we
update only when the envelope is first time being processed. This solves the issue in
Broadcast stream where a message need to be processed by multiple tasks. In that case,
the envelope will be in the pendingEnvelopeQueue of each task. Only the first fetch updates
the chooser with the next envelope in the broadcast stream partition.
The function will be called in the run loop thread so no synchronization.
@return
samza-core/src/main/java/org/apache/samza/container/RunLoop.java