Implementation:Apache Flink BatchCreator CreateNextBatch
| Knowledge Sources | |
|---|---|
| Domains | Stream_Processing, Async_IO |
| Last Updated | 2026-02-09 00:00 GMT |
Overview
Concrete tool for extracting sized batches from the request buffer and submitting them for asynchronous I/O provided by the Apache Flink connector-base module.
Description
The BatchCreator interface defines how entries are extracted from the buffer into batches. The default SimpleBatchCreator polls entries from the buffer up to the rate limiters allowed batch size, respecting both record count and byte size limits. The resulting Batch contains the entries list, total size in bytes, and record count. The batch is then passed to the abstract submitRequestEntries method which performs the actual async I/O. Results are reported via ResultHandler.
Usage
The default SimpleBatchCreator works for most use cases. Implement a custom BatchCreator only if non-standard batch formation logic is needed (e.g., grouping by partition key).
Code Reference
Source Location
- Repository: Apache Flink
- File: flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/BatchCreator.java
- Lines: L38-68
- File: flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/SimpleBatchCreator.java
- Lines: L40-79
Signature
public interface BatchCreator<RequestEntryT> {
Batch<RequestEntryT> createNextBatch(
RequestInfo requestInfo,
RequestBuffer<RequestEntryT> bufferedRequestEntries);
}
// Default implementation
public class SimpleBatchCreator<RequestEntryT> implements BatchCreator<RequestEntryT> {
public Batch<RequestEntryT> createNextBatch(
RequestInfo requestInfo,
RequestBuffer<RequestEntryT> bufferedRequestEntries);
}
Import
import org.apache.flink.connector.base.sink.writer.BatchCreator;
import org.apache.flink.connector.base.sink.writer.SimpleBatchCreator;
import org.apache.flink.connector.base.sink.writer.Batch;
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| requestInfo | RequestInfo | Yes | Contains getBatchSize() from rate limiter |
| bufferedRequestEntries | RequestBuffer<RequestEntryT> | Yes | Buffer to extract entries from |
Outputs
| Name | Type | Description |
|---|---|---|
| batch | Batch<RequestEntryT> | Contains entries list, total bytes, record count |
Usage Examples
Batch Formation
// SimpleBatchCreator.createNextBatch():
// 1. batchSize = min(requestInfo.getBatchSize(), buffer.size())
// 2. Poll up to batchSize entries from buffer
// 3. Stop if maxBatchSizeInBytes would be exceeded
// 4. Return Batch(entries, totalBytes, count)
//
// Then the writer calls:
// submitRequestEntries(batch.getBatchEntries(), resultHandler);