question: 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 ) ) ; } }
target:  Wait until enough data (key-value records) has been consumed.