@Override
    public void configureBulkProcessor(Builder builder, BulkProcessorOptions options) {
        builder.setConcurrentRequests(options.getConcurrentRequests());
        if (options.getFlushMaxActions() != null) {
            builder.setBulkActions(options.getFlushMaxActions());
        }
        if (options.getFlushMaxSizeMb() != null) {
            final ByteSizeUnit sizeUnit;
            if (options.getFlushMaxSizeMb() == -1) {
                // bulk size can be disabled with -1, however the ByteSizeValue constructor accepts -1
                // only with BYTES as the size unit
                sizeUnit = ByteSizeUnit.BYTES;
            } else {
                sizeUnit = ByteSizeUnit.MB;
            }
            builder.setBulkSize(new ByteSizeValue(options.getFlushMaxSizeMb(), sizeUnit));
        }
        if (options.getFlushInterval() != null) {
            if (options.getFlushInterval() == -1) {
                builder.setFlushInterval(null);
            } else {
                builder.setFlushInterval(TimeValue.timeValueMillis(options.getFlushInterval()));
            }
        }
    }
The factory to configure the rest client.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/Elasticsearch7ApiCallBridge.java