PartitionReader implements Comparable<PartitionReader> { public void setNextBatchSizeBytes(int nextBatchSizeBytes) { if (nextBatchSizeBytes < 0) return; this.m_nextBatchSizeBytes = nextBatchSizeBytes; } 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 testSetNextBatchSize() throws Exception { int nextBatchSizeBytes = (Integer) ReflectFieldUtil.getField(reader, "m_nextBatchSizeBytes"); assertEquals(consumerConfig.getBatchSizeBytes(), nextBatchSizeBytes); reader.setNextBatchSizeBytes(-100); nextBatchSizeBytes = (Integer) ReflectFieldUtil.getField(reader, "m_nextBatchSizeBytes"); assertEquals(consumerConfig.getBatchSizeBytes(), nextBatchSizeBytes); reader.setNextBatchSizeBytes(10000); nextBatchSizeBytes = (Integer) ReflectFieldUtil.getField(reader, "m_nextBatchSizeBytes"); assertTrue(consumerConfig.getBatchSizeBytes() != nextBatchSizeBytes); assertEquals(10000, nextBatchSizeBytes); }