PartitionReader implements Comparable<PartitionReader> { public void initOffset(boolean isNew) { if (LOGGER.isInfoEnabled()) LOGGER.info( "Init offset for " + m_clientId); if (isNew) { m_readOffset = fetchResetOffset(SmallestTimeString()); commitOffset(); } else { if (offsetExists()) m_readOffset = readZkOffset(); if (m_readOffset < 0) { m_readOffset = fetchResetOffset(m_config.getAutoOffsetReset()); commitOffset(); } } if (m_readOffset < 0) throw new RuntimeException( "Fatal errors in getting init offset for " + m_clientId); if (LOGGER.isInfoEnabled()) LOGGER.info( m_clientId + ": init offset is set to " + m_readOffset); } PartitionReader(String topic, int partition,
			KafkaConsumerConfig config,
			KafkaController.ZkConnector zkConnector,
			KafkaMessageSerializer serializer); void initOffset(boolean isNew); void resetOffset(); void revertOffset(); long getReadOffset(); void setReadOffset(long offset); void setNextBatchSizeBytes(int nextBatchSizeBytes); boolean isTaken(); void setToRelease(); void setLost(); boolean isLost(); String getTopic(); int getPartition(); String getId(); long calcWaitTime(); long getNextFetchInMs(); void setNextFetchInMs(long nextFetchInMs); void fistIdle(); long getIdleTime(); boolean isIdled(); void setOnIdled(); boolean isOnIdled(); void clearIdleStats(); List<JetstreamEvent> readEvents(); void commitOffset(); void close(); long readZkOffset(); @Override int compareTo(PartitionReader p);  }
@Test public void testInitOffset() throws Exception { long smallest = 100, largest = 10000; SimpleConsumer mockConsumer = mock(SimpleConsumer.class); mockSimpleConsumerForEarliestOffset(mockConsumer, topic, partition, smallest); mockSimpleConsumerForLatestOffset(mockConsumer, topic, partition, largest); ReflectFieldUtil.setField(reader, "m_consumer", mockConsumer); zkConnector.delete(offsetPath); reader.setReadOffset(-1); consumerConfig.setAutoOffsetReset("largest"); reader.initOffset(false); long readOffset = reader.getReadOffset(); assertEquals(largest, readOffset); zkConnector.writeString(offsetPath, String.valueOf(2000)); reader.setReadOffset(-1); consumerConfig.setAutoOffsetReset("largest"); reader.initOffset(false); readOffset = reader.getReadOffset(); assertEquals(2000, readOffset); zkConnector.delete(offsetPath); reader.setReadOffset(-1); consumerConfig.setAutoOffsetReset("smallest"); reader.initOffset(false); readOffset = reader.getReadOffset(); assertEquals(smallest, readOffset); zkConnector.delete(offsetPath); reader.setReadOffset(-1); consumerConfig.setAutoOffsetReset("largest"); reader.initOffset(true); readOffset = reader.getReadOffset(); assertEquals(smallest, readOffset); zkConnector.writeString(offsetPath, "offset"); boolean thrown = false; try { reader.initOffset(false); } catch (Exception e) { assertTrue(e instanceof RuntimeException); thrown = true; } assertTrue(thrown); }