@Test public void testBuildConsumer() throws Exception { String streamType = "kafka"; String streamKafkaTopicName = "theTopic"; String streamKafkaBrokerList = "127.0.0.1:" + kafkaCluster.getKafkaServerPort(0); String streamKafkaConsumerType = "simple"; String clientId = "clientId"; String tableNameWithType = "tableName_REALTIME"; 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", getKafkaConsumerFactoryName()); streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass"); streamConfigMap.put("stream.kafka.fetcher.size", "10000"); streamConfigMap.put("stream.kafka.fetcher.minBytes", "20000"); StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap); KafkaStreamMetadataProvider streamMetadataProvider = new KafkaStreamMetadataProvider(clientId, streamConfig); KafkaPartitionLevelConsumer kafkaSimpleStreamConsumer = new KafkaPartitionLevelConsumer(clientId, streamConfig, 0); kafkaSimpleStreamConsumer.fetchMessages(new LongMsgOffset(12345L), new LongMsgOffset(23456L), 10000); Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BUFFER_SIZE_DEFAULT, kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaBufferSize()); Assert.assertEquals(KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_SOCKET_TIMEOUT_DEFAULT, kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaSocketTimeout()); Assert.assertEquals(10000, kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaFetcherSizeBytes()); Assert .assertEquals(20000, kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaFetcherMinBytes()); streamConfigMap.put("stream.kafka.buffer.size", "100"); streamConfigMap.put("stream.kafka.socket.timeout", "1000"); streamConfig = new StreamConfig(tableNameWithType, streamConfigMap); kafkaSimpleStreamConsumer = new KafkaPartitionLevelConsumer(clientId, streamConfig, 0); kafkaSimpleStreamConsumer.fetchMessages(new LongMsgOffset(12345L), new LongMsgOffset(23456L), 10000); Assert.assertEquals(100, kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaBufferSize()); Assert.assertEquals(1000, kafkaSimpleStreamConsumer.getKafkaPartitionLevelStreamConfig().getKafkaSocketTimeout()); }
@Override public MessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, StreamPartitionMsgOffset endMsgOffset, int timeoutMillis) throws TimeoutException { final long startOffset = ((LongMsgOffset)startMsgOffset).getOffset(); final long endOffset = endMsgOffset == null ? Long.MAX_VALUE : ((LongMsgOffset)endMsgOffset).getOffset(); return fetchMessages(startOffset, endOffset, timeoutMillis); }
KafkaPartitionLevelConsumer extends KafkaPartitionLevelConnectionHandler implements PartitionLevelConsumer { @Override public MessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, StreamPartitionMsgOffset endMsgOffset, int timeoutMillis) throws TimeoutException { final long startOffset = ((LongMsgOffset)startMsgOffset).getOffset(); final long endOffset = endMsgOffset == null ? Long.MAX_VALUE : ((LongMsgOffset)endMsgOffset).getOffset(); return fetchMessages(startOffset, endOffset, timeoutMillis); } }
KafkaPartitionLevelConsumer extends KafkaPartitionLevelConnectionHandler implements PartitionLevelConsumer { @Override public MessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, StreamPartitionMsgOffset endMsgOffset, int timeoutMillis) throws 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); }
KafkaPartitionLevelConsumer extends KafkaPartitionLevelConnectionHandler implements PartitionLevelConsumer { @Override public MessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, StreamPartitionMsgOffset endMsgOffset, int timeoutMillis) throws 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); @Override MessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, StreamPartitionMsgOffset endMsgOffset,
      int timeoutMillis); MessageBatch fetchMessages(long startOffset, long endOffset, int timeoutMillis); @Override void close(); }
KafkaPartitionLevelConsumer extends KafkaPartitionLevelConnectionHandler implements PartitionLevelConsumer { @Override public MessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, StreamPartitionMsgOffset endMsgOffset, int timeoutMillis) throws 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); @Override MessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, StreamPartitionMsgOffset endMsgOffset,
      int timeoutMillis); MessageBatch fetchMessages(long startOffset, long endOffset, int timeoutMillis); @Override void close();  }