title: No current assignment for partition occurs even after poll in Kafka
body: <p>I have Java 8 application working with Apache Kafka 2.11-0.10.1.0. I need to use the <code>seek</code> feature to <code>poll</code> old messages from partitions. However I faced an exception of <code>No current assignment for partition</code> which is occurred every time I am trying to <code>seekByOffset</code>. Here's my class which is responsible for <code>seek</code>ing topics to the specified timestamp:</p>
  
  <pre><code>import org.apache.kafka.clients.consumer.ConsumerRecords;
  import org.apache.kafka.clients.consumer.KafkaConsumer;
  import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
  import org.apache.kafka.common.PartitionInfo;
  import org.apache.kafka.common.TopicPartition;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.springframework.util.CollectionUtils;
  
  import java.time.Instant;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.function.Function;
  import java.util.regex.Pattern;
  import java.util.stream.Collectors;
  
  /**
   * The main purpose of this class is to move fetching point for each partition of the {@link KafkaConsumer}
   * to some offset which is determined either by timestamp or by offset number.
   */
  public class KafkaSeeker {
      public static final long APP_STARTUP_TIME = Instant.now().toEpochMilli();
  
      private final Logger LOGGER = LoggerFactory.getLogger(this.getClass());
      private final KafkaConsumer&lt;String, String&gt; kafkaConsumer;
      private ConsumerRecords&lt;String, String&gt; polledRecords;
  
      public KafkaSeeker(KafkaConsumer&lt;String, String&gt; kafkaConsumer) {
          this.kafkaConsumer = kafkaConsumer;
          this.polledRecords = new ConsumerRecords&lt;&gt;(Collections.emptyMap());
      }
  
      /**
       * For each assigned or subscribed topic {@link org.apache.kafka.clients.consumer.KafkaConsumer#seek(TopicPartition, long)}
       * fetching pointer to the specified {@code timestamp}.
       * If no messages were found in each partition for a topic,
       * then {@link org.apache.kafka.clients.consumer.KafkaConsumer#seekToEnd(Collection)} will be called.
       *
       * Due to {@link KafkaConsumer#subscribe(Pattern)} and {@link KafkaConsumer#assign(Collection)} laziness
       * method needs to execute dummy {@link KafkaConsumer#poll(long)} method. All {@link ConsumerRecords} which were
       * polled from buffer are swallowed and produce warning logs.
       *
       * @param timestamp is used to find proper offset to seek to
       * @param topics are used to seek only specific topics. If not specified or empty, all subscribed topics are used.
       */
      public Map&lt;TopicPartition, OffsetAndTimestamp&gt; seek(long timestamp, Collection&lt;String&gt; topics) {
          this.polledRecords = kafkaConsumer.poll(0);
          Collection&lt;TopicPartition&gt; topicPartitions;
          if (CollectionUtils.isEmpty(topics)) {
              topicPartitions = kafkaConsumer.assignment();
          } else {
              topicPartitions = topics.stream()
                      .map(it -&gt; {
                          List&lt;Integer&gt; partitions = kafkaConsumer.partitionsFor(it).stream()
                                  .map(PartitionInfo::partition).collect(Collectors.toList());
                          return partitions.stream().map(partition -&gt; new TopicPartition(it, partition));
                      })
                      .flatMap(it -&gt; it)
                      .collect(Collectors.toList());
          }
  
          if (topicPartitions.isEmpty()) {
              throw new IllegalStateException("Kafka consumer doesn't have any subscribed topics.");
          }
  
          Map&lt;TopicPartition, Long&gt; timestampsByTopicPartitions = topicPartitions.stream()
                  .collect(Collectors.toMap(Function.identity(), topicPartition -&gt; timestamp));
          Map&lt;TopicPartition, Long&gt; beginningOffsets = kafkaConsumer.beginningOffsets(topicPartitions);
          Map&lt;TopicPartition, OffsetAndTimestamp&gt; offsets = kafkaConsumer.offsetsForTimes(timestampsByTopicPartitions);
          for (Map.Entry&lt;TopicPartition, OffsetAndTimestamp&gt; entry : offsets.entrySet()) {
              TopicPartition topicPartition = entry.getKey();
              if (entry.getValue() != null) {
                  LOGGER.info("Kafka seek topic:partition [{}:{}] from [{} offset] to [{} offset].",
                          topicPartition.topic(),
                          topicPartition.partition(),
                          beginningOffsets.get(topicPartition),
                          entry.getValue());
                  kafkaConsumer.seek(topicPartition, entry.getValue().offset());
              } else {
                  LOGGER.info("Kafka seek topic:partition [{}:{}] from [{} offset] to the end of partition.",
                          topicPartition.topic(),
                          topicPartition.partition());
                  kafkaConsumer.seekToEnd(Collections.singleton(topicPartition));
              }
          }
          return offsets;
      }
  
      public ConsumerRecords&lt;String, String&gt; getPolledRecords() {
          return polledRecords;
      }
  }
  </code></pre>
  
  <p>Before calling the method I have consumer subscribed to a single topic like this <code>consumer.subscribe(singletonList(kafkaTopic));</code>. When I get <code>kafkaConsumer.assignment()</code> it returns zero <code>TopicPartition</code>s assigned.  But if I specify the topic and get its partitions then  I have valid <code>TopicPartition</code>s, although they are failing on <code>seek</code> call with the error in the title. What is something I forgot?</p>