@Test public void testSendKafkaMessage() throws InterruptedException { logger.log(Level.INFO, "sendKafkaMessage"); logger.log(Level.INFO, "Initialising producer"); KafkaUtils instance = new KafkaUtils(TestUtilsForKafka.getConsumerProperties(BROKERHOST, BROKERPORT), TestUtilsForKafka.getProducerProperties("rcv-test", BROKERHOST, BROKERPORT), CONSUMER_TIMEOUT); instance.initializeProducer(); logger.log(Level.INFO, "Initialising consumer"); KafkaConsumer<String, byte[]> consumer; consumer = new KafkaConsumer<>(TestUtilsForKafka.getConsumerProperties(BROKERHOST, BROKERPORT)); consumer.subscribe(Arrays.asList(TOPIC_S)); logger.log(Level.INFO, "Produce data"); List<byte[]> sent = new ArrayList<>(); Random r = new Random(); InstancesHeader header = TestUtilsForKafka.generateHeader(10); Gson gson = new Gson(); for (int i = 0; i < NUM_INSTANCES; i++) { byte[] val = gson.toJson(TestUtilsForKafka.getData(r, 10, header)).getBytes(); sent.add(val); instance.sendKafkaMessage(TOPIC_S, val); } Thread.sleep(2 * CONSUMER_TIMEOUT); logger.log(Level.INFO, "Get results from Kafka"); List<byte[]> consumed = new ArrayList<>(); while (consumed.size() != sent.size()) { ConsumerRecords<String, byte[]> records = consumer.poll(CONSUMER_TIMEOUT); Iterator<ConsumerRecord<String, byte[]>> it = records.iterator(); while (it.hasNext()) { consumed.add(it.next().value()); } } consumer.close(); assertArrayEquals(sent.toArray(), consumed.toArray()); }
public long sendKafkaMessage(String topic, byte[] message) { if (producer != null) { try{ ProducerRecord<String, byte[]> record = new ProducerRecord(topic, message); long offset = producer.send(record).get(10, TimeUnit.SECONDS).offset(); producer.flush(); return offset; } catch(InterruptedException | ExecutionException | TimeoutException e){ Logger.getLogger(KafkaUtils.class.getName()).log(Level.SEVERE, null, e); } } return -1; }
KafkaUtils { public long sendKafkaMessage(String topic, byte[] message) { if (producer != null) { try{ ProducerRecord<String, byte[]> record = new ProducerRecord(topic, message); long offset = producer.send(record).get(10, TimeUnit.SECONDS).offset(); producer.flush(); return offset; } catch(InterruptedException | ExecutionException | TimeoutException e){ Logger.getLogger(KafkaUtils.class.getName()).log(Level.SEVERE, null, e); } } return -1; } }
KafkaUtils { public long sendKafkaMessage(String topic, byte[] message) { if (producer != null) { try{ ProducerRecord<String, byte[]> record = new ProducerRecord(topic, message); long offset = producer.send(record).get(10, TimeUnit.SECONDS).offset(); producer.flush(); return offset; } catch(InterruptedException | ExecutionException | TimeoutException e){ Logger.getLogger(KafkaUtils.class.getName()).log(Level.SEVERE, null, e); } } return -1; } KafkaUtils(Properties consumerProperties, Properties producerProperties, long consumerTimeout);   KafkaUtils(KafkaUtils kafkaUtils); }
KafkaUtils { public long sendKafkaMessage(String topic, byte[] message) { if (producer != null) { try{ ProducerRecord<String, byte[]> record = new ProducerRecord(topic, message); long offset = producer.send(record).get(10, TimeUnit.SECONDS).offset(); producer.flush(); return offset; } catch(InterruptedException | ExecutionException | TimeoutException e){ Logger.getLogger(KafkaUtils.class.getName()).log(Level.SEVERE, null, e); } } return -1; } 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 long sendKafkaMessage(String topic, byte[] message) { if (producer != null) { try{ ProducerRecord<String, byte[]> record = new ProducerRecord(topic, message); long offset = producer.send(record).get(10, TimeUnit.SECONDS).offset(); producer.flush(); return offset; } catch(InterruptedException | ExecutionException | TimeoutException e){ Logger.getLogger(KafkaUtils.class.getName()).log(Level.SEVERE, null, e); } } return -1; } 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);  }