protected static void putRecordBatchIntoDeliveryStream ( ) throws IOException { try ( InputStream inputStream = Thread . currentThread ( ) . getContextClassLoader ( ) . getResourceAsStream ( BATCH_PUT_STREAM_SOURCE ) ) { if ( inputStream == null ) { throw new FileNotFoundException ( "Could not find file " + BATCH_PUT_STREAM_SOURCE ) ; } List < Record > recordList = new ArrayList < Record > ( ) ; int batchSize = 0 ; try ( BufferedReader reader = new BufferedReader ( new InputStreamReader ( inputStream ) ) ) { String line = null ; while ( ( line = reader . readLine ( ) ) != null ) { String data = line + "\n" ; Record record = createRecord ( data ) ; recordList . add ( record ) ; batchSize ++ ; if ( batchSize == BATCH_PUT_MAX_SIZE ) { putRecordBatch ( recordList ) ; recordList . clear ( ) ; batchSize = 0 ; } } if ( batchSize > 0 ) { putRecordBatch ( recordList ) ; } } } }
Method to put records in the specified delivery stream by reading contents from sample input file using PutRecordBatch API .