PartitionReader implements Comparable<PartitionReader> { public void commitOffset() { if (!m_zkConnector.exists(m_offsetPath)) m_zkConnector.create(m_offsetPath, false); m_zkConnector.writeString(m_offsetPath, String.valueOf(m_readOffset)); if (LOGGER.isInfoEnabled()) LOGGER.info( m_clientId + ": offset is committed with " + 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 testCommitOffset() { zkConnector.delete(offsetPath); reader.setReadOffset(12345); reader.commitOffset(); reader.setReadOffset(0); reader.revertOffset(); assertEquals(12345, reader.getReadOffset()); }