private void fetchMessages() {
    Set<SystemStreamPartition> sspsToFetch = new HashSet<>();
    for (SystemStreamPartition ssp : nextOffsets.keySet()) {
      if (sink.needsMoreMessages(ssp)) {
        sspsToFetch.add(ssp);
      }
    }
    LOG.debug("pollConsumer for {} SSPs: {}", sspsToFetch.size(), sspsToFetch);
    if (!sspsToFetch.isEmpty()) {
      kafkaConsumerMetrics.incClientReads(metricName);

      Map<SystemStreamPartition, List<IncomingMessageEnvelope>> response;

      response = pollConsumer(sspsToFetch, 500L);

      // move the responses into the queue
      for (Map.Entry<SystemStreamPartition, List<IncomingMessageEnvelope>> e : response.entrySet()) {
        List<IncomingMessageEnvelope> envelopes = e.getValue();
        if (envelopes != null) {
          moveMessagesToTheirQueue(e.getKey(), envelopes);
        }
      }

      populateCurrentLags(sspsToFetch); // find current lags for for each SSP
    } else { // nothing to read

      LOG.debug("No topic/partitions need to be fetched for system {} right now. Sleeping {}ms.", systemName,
          SLEEP_MS_WHILE_NO_TOPIC_PARTITION);

      kafkaConsumerMetrics.incClientSkippedFetchRequests(metricName);

      try {
        Thread.sleep(SLEEP_MS_WHILE_NO_TOPIC_PARTITION);
      } catch (InterruptedException e) {
        LOG.warn("Sleep in fetchMessages was interrupted");
      }
    }
    refreshLagMetrics();
  }
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