private Map<SystemStreamPartition, List<IncomingMessageEnvelope>> pollConsumer(
      Set<SystemStreamPartition> systemStreamPartitions, long timeoutMs) {

    // Since we need to poll only from some subset of TopicPartitions (passed as the argument),
    // we need to pause the rest.
    List<TopicPartition> topicPartitionsToPause = new ArrayList<>();
    List<TopicPartition> topicPartitionsToPoll = new ArrayList<>();

    for (Map.Entry<TopicPartition, SystemStreamPartition> e : topicPartitionToSSP.entrySet()) {
      TopicPartition tp = e.getKey();
      SystemStreamPartition ssp = e.getValue();
      if (systemStreamPartitions.contains(ssp)) {
        topicPartitionsToPoll.add(tp);  // consume
      } else {
        topicPartitionsToPause.add(tp); // ignore
      }
    }

    ConsumerRecords<K, V> records;
    try {
      // Synchronize, in case the consumer is used in some other thread (metadata or something else)
      synchronized (kafkaConsumer) {
        // Since we are not polling from ALL the subscribed topics, so we need to "change" the subscription temporarily
        kafkaConsumer.pause(topicPartitionsToPause);
        kafkaConsumer.resume(topicPartitionsToPoll);
        records = kafkaConsumer.poll(timeoutMs);
      }
    } catch (Exception e) {
      // we may get InvalidOffsetException | AuthorizationException | KafkaException exceptions,
      // but we still just rethrow, and log it up the stack.
      LOG.error("Caught a Kafka exception in pollConsumer for system " + systemName, e);
      throw e;
    }

    return processResults(records);
  }
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