void addMessage(SystemStreamPartition ssp, IncomingMessageEnvelope envelope) {
      LOG.trace("{}: Incoming message ssp = {}: envelope = {}.", this, ssp, envelope);

      try {
        put(ssp, envelope);
      } catch (InterruptedException e) {
        throw new SamzaException(
            String.format("%s: Consumer was interrupted while trying to add message with offset %s for ssp %s", this,
                envelope.getOffset(), ssp));
      }
    }
return system name for this consumer
@return system name
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemConsumer.java