public < K , V > ListenableFuture < List < ConsumerRecord < K , V > > > consume ( String topic , KafkaConsumer < K , V > consumer , int numMessagesToConsume ) { consumer . subscribe ( Lists . newArrayList ( topic ) ) ; ListeningExecutorService executor = MoreExecutors . listeningDecorator ( Executors . newSingleThreadExecutor ( ) ) ; return executor . submit ( new RecordConsumer <> ( numMessagesToConsume , consumer ) ) ; }
Attempt to consume the specified number of messages