public static <K,V>List<KeyValue<K,V>> waitUntilMinKeyValueRecordsReceived(Properties consumerConfig,String topic,int expectedNumRecords,long waitTime) throws InterruptedException {
  List<KeyValue<K,V>> accumData=new ArrayList<>();
  long startTime=System.currentTimeMillis();
  while (true) {
    List<KeyValue<K,V>> readData=readKeyValues(topic,consumerConfig);
    accumData.addAll(readData);
    if (accumData.size() >= expectedNumRecords)     return accumData;
    if (System.currentTimeMillis() > startTime + waitTime)     throw new AssertionError("Expected " + expectedNumRecords + " but received only "+ accumData.size()+ " records before timeout "+ waitTime+ " ms");
    Thread.sleep(Math.min(waitTime,100L));
  }
}

public static < K , V > List < KeyValue < K , V > > waitUntilMinKeyValueRecordsReceived ( Properties consumerConfig , String topic , int expectedNumRecords , long waitTime ) throws InterruptedException { List < KeyValue < K , V > > accumData = new ArrayList < > ( ) ; long startTime = System . currentTimeMillis ( ) ; while ( true ) { List < KeyValue < K , V > > readData = readKeyValues ( topic , consumerConfig ) ; accumData . addAll ( readData ) ; if ( accumData . size ( ) >= expectedNumRecords ) return accumData ; if ( System . currentTimeMillis ( ) > startTime + waitTime ) throw new AssertionError ( "Expected " + expectedNumRecords + " but received only " + accumData . size ( ) + " records before timeout " + waitTime + " ms" ) ; Thread . sleep ( Math . min ( waitTime , 100L ) ) ; } }
wait until enough data ( key - value records ) has been consumed .
