title: How to find the offset range for a topic-partition in Kafka 0.10?
body: <p>I'm using Kafka 0.10.0. Before processing, I want to know the size of the records in a partition.</p>
  
  <p>In 0.9.0.1 version, I used to find the difference between <code>latest</code> and <code>earliest</code> offset for a partition by using the below code. In the new version, it gets stuck when retrieving the <code>consumer#position</code> method.</p>
  
  <pre><code>package org.apache.kafka.example.utils;
  
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  
  import org.apache.commons.lang3.Range;
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.clients.consumer.KafkaConsumer;
  import org.apache.kafka.common.PartitionInfo;
  import org.apache.kafka.common.TopicPartition;
  import org.apache.kafka.common.serialization.ByteArraySerializer;
  import org.apache.logging.log4j.LogManager;
  import org.apache.logging.log4j.Logger;
  
  public class FindTopicRange {
  
      private static Logger logger = LogManager.getLogger();
  
      public FindTopicRange() {
          // TODO Auto-generated constructor stub
      }
  
      public static Map&lt;TopicPartition, Range&lt;Long&gt;&gt; getOffsets(String topic) {
  
          Map&lt;TopicPartition, Range&lt;Long&gt;&gt; partitionToRange = new HashMap&lt;&gt;();
          try (KafkaConsumer&lt;byte[], byte[]&gt; consumer = new KafkaConsumer&lt;&gt;(getConsumerConfigs())) {
  
              List&lt;TopicPartition&gt; partitions = new ArrayList&lt;&gt;();
              for (PartitionInfo partitionInfo : consumer.partitionsFor(topic)) {
                  partitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition()));
              }
              consumer.assign(partitions);
  
              for (TopicPartition partition : partitions) {
                  consumer.seekToBeginning(Collections.singletonList(partition));
                  long earliestOffset = consumer.position(partition);
  
                  consumer.seekToEnd(Collections.singletonList(partition));
                  long latestOffset = consumer.position(partition);
                  partitionToRange.put(partition, Range.between(earliestOffset, latestOffset));
              }
              return partitionToRange;
          } catch (Exception e) {
              logger.error("Exception while getting offset range information for topic : {}", topic, e);
          }
          return partitionToRange;
      }
  
      private static Properties getConsumerConfigs() {
          Properties configs = new Properties();
          configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
          configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
          configs.put(ConsumerConfig.CLIENT_ID_CONFIG, "test");
          configs.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 10240);
  
          configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getCanonicalName());
          configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getCanonicalName());
          return configs;
      }
  
      public static void main(String[] args) {
          System.out.println(getOffsets("hello"));
      }
  
  }
  </code></pre>
  
  <p>Stacktrace for the above call is shown below:</p>
  
  <pre><code>"main" prio=10 tid=0x00007f1750013800 nid=0x443 runnable [0x00007f1756b88000]
     java.lang.Thread.State: RUNNABLE
          at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
          at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
          at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:79)
          at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:87)
          - locked &lt;0x00000007c21cba00&gt; (a sun.nio.ch.Util$2)
          - locked &lt;0x00000007c21cb9f0&gt; (a java.util.Collections$UnmodifiableSet)
          - locked &lt;0x00000007c21cb8d8&gt; (a sun.nio.ch.EPollSelectorImpl)
          at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:98)
          at org.apache.kafka.common.network.Selector.select(Selector.java:454)
          at org.apache.kafka.common.network.Selector.poll(Selector.java:277)
          at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:360)
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224)
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:192)
          at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:134)
          at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:183)
          at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.fetchCommittedOffsets(ConsumerCoordinator.java:324)
          at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.refreshCommittedOffsetsIfNeeded(ConsumerCoordinator.java:306)
          at org.apache.kafka.clients.consumer.KafkaConsumer.updateFetchPositions(KafkaConsumer.java:1405)
          at org.apache.kafka.clients.consumer.KafkaConsumer.position(KafkaConsumer.java:1196)
  </code></pre>