private IncomingMessageEnvelope chooseEnvelope() {
    IncomingMessageEnvelope envelope = consumerMultiplexer.choose(false);
    if (envelope != null) {
      log.trace("Choose envelope ssp {} offset {} for processing",
          envelope.getSystemStreamPartition(elasticityFactor), envelope.getOffset());
      containerMetrics.envelopes().inc();
    } else {
      log.trace("No envelope is available");
      containerMetrics.nullEnvelopes().inc();
    }
    return envelope;
  }
Chooses an envelope from messageChooser without updating it. This enables flow control
on the SSP level, meaning the task will not get further messages for the SSP if it cannot
process it. The chooser is updated only after the callback to process is invoked, then the task
is able to process more messages. This flow control does not block. so in case of empty message chooser,
it will return null immediately without blocking, and the chooser will not poll the underlying system
consumer since there are still messages in the SystemConsumers buffer.
samza-core/src/main/java/org/apache/samza/container/RunLoop.java