@Test public void testInitializeConsumer() throws Exception { logger.log(Level.INFO, "initializeConsumer"); Collection<String> topics = Arrays.asList(TOPIC_R); KafkaUtils instance = new KafkaUtils(TestUtilsForKafka.getConsumerProperties(BROKERHOST, BROKERPORT), TestUtilsForKafka.getProducerProperties(BROKERHOST, BROKERPORT), CONSUMER_TIMEOUT); assertNotNull(instance); instance.initializeConsumer(topics); Thread.sleep(1000); instance.closeConsumer(); Thread.sleep(CONSUMER_TIMEOUT); instance.initializeConsumer(topics); Thread.sleep(1000); instance.closeConsumer(); assertTrue(true); }
public void initializeConsumer(Collection<String> topics) { kafkaConsumerThread = new KafkaConsumerThread(consumerProperties, topics, consumerTimeout); kafkaConsumerThread.start(); }
KafkaUtils { public void initializeConsumer(Collection<String> topics) { kafkaConsumerThread = new KafkaConsumerThread(consumerProperties, topics, consumerTimeout); kafkaConsumerThread.start(); } }
KafkaUtils { public void initializeConsumer(Collection<String> topics) { kafkaConsumerThread = new KafkaConsumerThread(consumerProperties, topics, consumerTimeout); kafkaConsumerThread.start(); } KafkaUtils(Properties consumerProperties, Properties producerProperties, long consumerTimeout);   KafkaUtils(KafkaUtils kafkaUtils); }
KafkaUtils { public void initializeConsumer(Collection<String> topics) { kafkaConsumerThread = new KafkaConsumerThread(consumerProperties, topics, consumerTimeout); kafkaConsumerThread.start(); } 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 void initializeConsumer(Collection<String> topics) { kafkaConsumerThread = new KafkaConsumerThread(consumerProperties, topics, consumerTimeout); kafkaConsumerThread.start(); } 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);  }