private List<Object> addToPendingSendBatch(int targetTask, String streamId, List<Object> values) {
        Map<String, List<Object>> streamToBatch = pendingSendMsgs.get(targetTask);
        if (streamToBatch == null) {
            streamToBatch = new HashMap<>();
            pendingSendMsgs.put(targetTask, streamToBatch);
        }

        List<Object> batch = streamToBatch.get(streamId);
        if (batch == null) {
            batch = new ArrayList<>();
            streamToBatch.put(streamId, batch);
        }

        batch.addAll(values);
        if (batch.size() >= batchSize) {
            return batch;
        } else {
            return null;
        }
    }
private List < Object > addToPendingSendBatch ( int targetTask , String streamId , List < Object > values ) { Map < String , List < Object > > streamToBatch = pendingSendMsgs . get ( targetTask ) ; if ( streamToBatch == null ) { streamToBatch = new HashMap < > ( ) ; pendingSendMsgs . put ( targetTask , streamToBatch ) ; } List < Object > batch = streamToBatch . get ( streamId ) ; if ( batch == null ) { batch = new ArrayList < > ( ) ; streamToBatch . put ( streamId , batch ) ; } batch . addAll ( values ) ; if ( batch . size ( ) >= batchSize ) { return batch ; } else { return null ; } }
@return if size of pending batch is bigger than the configured one, return the batch for sending, otherwise, return null