PartitionReader implements Comparable<PartitionReader> { public void resetOffset() { if (LOGGER.isInfoEnabled()) LOGGER.info( "Reset offset for " + m_clientId); m_readOffset = fetchResetOffset(m_config.getAutoOffsetReset()); if (m_readOffset < 0) throw new RuntimeException("Fatal errors in resetting offset for " + m_clientId); commitOffset(); if (LOGGER.isInfoEnabled()) LOGGER.info( m_clientId + ": offset is reset 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 testResetOffset() 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); reader.resetOffset(); long readOffset = reader.getReadOffset(); assertEquals(largest, readOffset); consumerConfig.setAutoOffsetReset("smallest"); reader.resetOffset(); readOffset = reader.getReadOffset(); assertEquals(smallest, readOffset); }
@Test public void testOffsetError() throws Exception { SimpleConsumer mockConsumer = mock(SimpleConsumer.class); mockSimpleConsumerForOffsetError(mockConsumer, topic, partition); ReflectFieldUtil.setField(reader, "m_consumer", mockConsumer); try { reader.resetOffset(); } catch (Exception e) { } verify(mockConsumer, atLeast(2)).getOffsetsBefore( argThat(new IsOffsetRequest())); }