private Map<String,String> produceStream(String topic){
  List<Integer> subnet=new ArrayList<>();
  for (int i=1; i <= CNT; i++)   subnet.add(i);
  Collections.shuffle(subnet);
  List<ProducerRecord<String,String>> messages=new ArrayList<>(CNT);
  Map<String,String> keyValMap=new HashMap<>();
  for (int evt=0; evt < CNT; evt++) {
    long runtime=System.currentTimeMillis();
    String ip=KEY_PREFIX + subnet.get(evt);
    String msg=runtime + VALUE_URL + ip;
    messages.add(new ProducerRecord<>(topic,ip,msg));
    keyValMap.put(ip,msg);
  }
  embeddedBroker.sendMessages(messages);
  return keyValMap;
}

private Map < String , String > produceStream ( String topic ) { List < Integer > subnet = new ArrayList < > ( ) ; for ( int i = 1 ; i <= CNT ; i ++ ) subnet . add ( i ) ; Collections . shuffle ( subnet ) ; List < ProducerRecord < String , String > > messages = new ArrayList < > ( CNT ) ; Map < String , String > keyValMap = new HashMap < > ( ) ; for ( int evt = 0 ; evt < CNT ; evt ++ ) { long runtime = System . currentTimeMillis ( ) ; String ip = KEY_PREFIX + subnet . get ( evt ) ; String msg = runtime + VALUE_URL + ip ; messages . add ( new ProducerRecord < > ( topic , ip , msg ) ) ; keyValMap . put ( ip , msg ) ; } embeddedBroker . sendMessages ( messages ) ; return keyValMap ; }
sends messages to kafka .
