private Map<SystemStreamPartition, List<IncomingMessageEnvelope>> processResults(ConsumerRecords<K, V> records) {
    if (records == null) {
      throw new SamzaException("Received null 'records' after polling consumer in KafkaConsumerProxy " + this);
    }

    Map<SystemStreamPartition, List<IncomingMessageEnvelope>> results = new HashMap<>(records.count());
    // Parse the returned records and convert them into the IncomingMessageEnvelope.
    for (ConsumerRecord<K, V> record : records) {
      int partition = record.partition();
      String topic = record.topic();
      TopicPartition tp = new TopicPartition(topic, partition);

      updateMetrics(record, tp);

      SystemStreamPartition ssp = topicPartitionToSSP.get(tp);
      List<IncomingMessageEnvelope> messages = results.computeIfAbsent(ssp, k -> new ArrayList<>());

      IncomingMessageEnvelope incomingMessageEnvelope = handleNewRecord(record, ssp);
      messages.add(incomingMessageEnvelope);
    }
    if (LOG.isDebugEnabled()) {
      LOG.debug("# records per SSP:");
      for (Map.Entry<SystemStreamPartition, List<IncomingMessageEnvelope>> e : results.entrySet()) {
        List<IncomingMessageEnvelope> list = e.getValue();
        LOG.debug(e.getKey() + " = " + ((list == null) ? 0 : list.size()));
      }
    }

    return results;
  }
Stop this KafkaConsumerProxy and wait for at most {@code timeoutMs}.
@param timeoutMs maximum time to wait to stop this KafkaConsumerProxy
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaConsumerProxy.java