@Test public void testClose() throws Exception { SimpleConsumer mockConsumer = mock(SimpleConsumer.class); ReflectFieldUtil.setField(reader, "m_consumer", mockConsumer); reader.close(); verify(mockConsumer).close(); }
public void close() { m_consumer.close(); }
PartitionReader implements Comparable<PartitionReader> { public void close() { m_consumer.close(); } }
PartitionReader implements Comparable<PartitionReader> { public void close() { m_consumer.close(); } PartitionReader(String topic, int partition,
			KafkaConsumerConfig config,
			KafkaController.ZkConnector zkConnector,
			KafkaMessageSerializer serializer); }
PartitionReader implements Comparable<PartitionReader> { public void close() { m_consumer.close(); } 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); }
PartitionReader implements Comparable<PartitionReader> { public void close() { m_consumer.close(); } 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);  }