public ListenableFuture < List < String > > consumeStrings ( String topic , int numMessagesToConsume ) { KafkaConsumer < String , String > consumer = createStringConsumer ( ) ; ListenableFuture < List < ConsumerRecord < String , String > > > records = consume ( topic , consumer , numMessagesToConsume ) ; return Futures . transform ( records , this :: extractValues , MoreExecutors . directExecutor ( ) ) ; }
Consume specified number of string messages