@VisibleForTesting
    protected BulkProcessor buildBulkProcessor(Listener listener) {
        checkNotNull(listener);
        Builder builder = callBridge.createBulkProcessorBuilder(client, listener);
        // if backoff retrying is disabled, bulkProcessorFlushBackoffPolicy will be null
        callBridge.configureBulkProcessorBackoff(
                builder, bulkProcessorFlushBackoffPolicy);
        /// This makes flush() blocking
        // builder.setConcurrentRequests(0);
        callBridge.configureBulkProcessor(builder, new BulkProcessorOptions(bulkProcessorFlushMaxActions,
                bulkProcessorFlushMaxSizeMb, bulkProcessorFlushIntervalMillis, 0));
        return callBridge.buildBulkProcessor(builder);
    }
Build the {@link BulkProcessor}.

<p>Note: this is exposed for testing purposes.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/ElasticsearchSinkBase.java