@Test public void testGetKafkaMessages() throws Exception { logger.log(Level.INFO, "getKafkaMessages"); Collection<String> topics = Arrays.asList(TOPIC_R); KafkaUtils instance = new KafkaUtils(TestUtilsForKafka.getConsumerProperties(BROKERHOST, BROKERPORT), TestUtilsForKafka.getProducerProperties(BROKERHOST, BROKERPORT), CONSUMER_TIMEOUT); assertNotNull(instance); logger.log(Level.INFO, "Initialising consumer"); instance.initializeConsumer(topics); logger.log(Level.INFO, "Produce data"); List expResult = sendAndGetMessages(NUM_INSTANCES); logger.log(Level.INFO, "Wait a moment"); Thread.sleep(CONSUMER_TIMEOUT); logger.log(Level.INFO, "Get results from Kafka"); List<byte[]> result = instance.getKafkaMessages(); assertArrayEquals(expResult.toArray(), result.toArray()); instance.closeConsumer(); }
public List<byte[]> getKafkaMessages() throws Exception { return kafkaConsumerThread.getKafkaMessages(); }
KafkaUtils { public List<byte[]> getKafkaMessages() throws Exception { return kafkaConsumerThread.getKafkaMessages(); } }
KafkaUtils { public List<byte[]> getKafkaMessages() throws Exception { return kafkaConsumerThread.getKafkaMessages(); } KafkaUtils(Properties consumerProperties, Properties producerProperties, long consumerTimeout);   KafkaUtils(KafkaUtils kafkaUtils); }
KafkaUtils { public List<byte[]> getKafkaMessages() throws Exception { return kafkaConsumerThread.getKafkaMessages(); } KafkaUtils(Properties consumerProperties, Properties producerProperties, long consumerTimeout);   KafkaUtils(KafkaUtils kafkaUtils); void initializeConsumer(Collection<String> topics); void closeConsumer(); void initializeProducer(); void closeProducer(); List<byte[]> getKafkaMessages(); long sendKafkaMessage(String topic, byte[] message); }
KafkaUtils { public List<byte[]> getKafkaMessages() throws Exception { return kafkaConsumerThread.getKafkaMessages(); } KafkaUtils(Properties consumerProperties, Properties producerProperties, long consumerTimeout);   KafkaUtils(KafkaUtils kafkaUtils); void initializeConsumer(Collection<String> topics); void closeConsumer(); void initializeProducer(); void closeProducer(); List<byte[]> getKafkaMessages(); long sendKafkaMessage(String topic, byte[] message);  }