### Question:
KafkaPartitionLevelConsumer extends KafkaConnectionHandler implements PartitionLevelConsumer { @Override public synchronized MessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, StreamPartitionMsgOffset endMsgOffset, int timeoutMillis) throws java.util.concurrent.TimeoutException { final long startOffset = ((LongMsgOffset) startMsgOffset).getOffset(); final long endOffset = (endMsgOffset == null) ? Long.MAX_VALUE : ((LongMsgOffset) endMsgOffset).getOffset(); return fetchMessages(startOffset, endOffset, timeoutMillis); } KafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition); @VisibleForTesting  KafkaPartitionLevelConsumer(String clientId, StreamConfig streamConfig, int partition,
      KafkaSimpleConsumerFactory kafkaSimpleConsumerFactory); @Override synchronized MessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset,
      StreamPartitionMsgOffset endMsgOffset, int timeoutMillis); synchronized MessageBatch fetchMessages(long startOffset, long endOffset, int timeoutMillis); @VisibleForTesting int getFetchRequestSize(); @VisibleForTesting int getFetchRequestMinBytes(); @Override /** * Closes this consumer. */ void close();  }### Answer:
@Test public void testFetchMessages() throws Exception { String streamType = "kafka"; String streamKafkaTopicName = "theTopic"; String streamKafkaBrokerList = "abcd:1234,bcde:2345"; String streamKafkaConsumerType = "simple"; String clientId = "clientId"; String tableNameWithType = "table_REALTIME"; MockKafkaSimpleConsumerFactory mockKafkaSimpleConsumerFactory = new MockKafkaSimpleConsumerFactory(new String[]{"abcd", "bcde"}, new int[]{1234, 2345}, new long[]{12345L, 23456L}, new long[]{23456L, 34567L}, new int[]{0, 1}, streamKafkaTopicName); Map<String, String> streamConfigMap = new HashMap<>(); streamConfigMap.put("streamType", streamType); streamConfigMap.put("stream.kafka.topic.name", streamKafkaTopicName); streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList); streamConfigMap.put("stream.kafka.consumer.type", streamKafkaConsumerType); streamConfigMap .put("stream.kafka.consumer.factory.class.name", mockKafkaSimpleConsumerFactory.getClass().getName()); streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass"); StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap); int partition = 0; KafkaPartitionLevelConsumer kafkaSimpleStreamConsumer = new KafkaPartitionLevelConsumer(clientId, streamConfig, partition, mockKafkaSimpleConsumerFactory); kafkaSimpleStreamConsumer.fetchMessages(new LongMsgOffset(12345L), new LongMsgOffset(23456L), 10000); }