PartitionReader implements Comparable<PartitionReader> { public void revertOffset() { long oldOffset = m_readOffset; m_readOffset = readZkOffset(); if (m_readOffset < 0) { LOGGER.warn( m_clientId + "No offset node in zookeeper."); resetOffset(); } if (LOGGER.isInfoEnabled()) LOGGER.info( m_clientId + ": offset is reverted from " + oldOffset + " 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 testRevertOffset() 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.create(offsetPath, false); zkConnector.writeString(offsetPath, String.valueOf(2000)); reader.setReadOffset(3000); assertEquals(3000, reader.getReadOffset()); reader.revertOffset(); assertEquals(2000, reader.getReadOffset()); zkConnector.writeString(offsetPath, String.valueOf(-2)); reader.setReadOffset(3000); assertEquals(3000, reader.getReadOffset()); consumerConfig.setAutoOffsetReset("largest"); reader.revertOffset(); assertEquals(largest, reader.getReadOffset()); }