private List<AsyncTaskWorker> getWorkersForEnvelope(IncomingMessageEnvelope envelope) {
    if (elasticityFactor <= 1) {
      return sspToTaskWorkerMapping.get(envelope.getSystemStreamPartition());
    }

    final SystemStreamPartition sspOfEnvelope = envelope.getSystemStreamPartition(elasticityFactor);
    List<AsyncTaskWorker> listOfWorkersForEnvelope = null;

    // if envelope is end of stream or watermark or drain, it needs to be routed to all tasks consuming the ssp irresp
    // of keybucket
    MessageType messageType = MessageType.of(envelope.getMessage());
    if (envelope.isEndOfStream()
        || envelope.isDrain()
        || messageType == MessageType.END_OF_STREAM
        || messageType == MessageType.DRAIN
        || messageType == MessageType.WATERMARK) {

      //sspToTaskWorkerMapping has ssps with keybucket so extract and check only system, stream and partition and ignore the keybucket
      listOfWorkersForEnvelope = sspToTaskWorkerMapping.entrySet()
          .stream()
          .filter(sspToTask -> sspToTask.getKey().getSystemStream().equals(sspOfEnvelope.getSystemStream())
              && sspToTask.getKey().getPartition().equals(sspOfEnvelope.getPartition()))
          .map(sspToWorker -> sspToWorker.getValue())
          .flatMap(Collection::stream)
          .collect(Collectors.toList());
    } else {
      listOfWorkersForEnvelope = sspToTaskWorkerMapping.get(sspOfEnvelope);
    }
    return listOfWorkersForEnvelope;
  }
when elasticity is not enabled, fetch the workers from sspToTaskWorkerMapping using envelope.getSSP()
when elasticity is enabled,
      sspToTaskWorkerMapping has workers for a SSP which has keyBucket
      hence need to use envelop.getSSP(elasticityFactor)
      Additionally, when envelope is EndOfStream or Watermark or Drain, it needs to be sent to all works for the ssp
      irrespective of keyBucket
@param envelope
@return list of workers for the envelope
samza-core/src/main/java/org/apache/samza/container/RunLoop.java