{"record":{"id":"952f47c9a7f20726","repo":"apache/flink","slug":"the-request-entry-sent-to-the-buffer-was-of-size","errorCode":null,"errorMessage":"The request entry sent to the buffer was of size [%s], when the maxRecordSizeInBytes was set to [%s].","messagePattern":"The request entry sent to the buffer was of size \\[(.+?)\\], when the maxRecordSizeInBytes was set to \\[(.+?)\\]\\.","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java","lineNumber":416,"sourceCode":"        ListIterator<RequestEntryT> iterator =\n                failedRequestEntries.listIterator(failedRequestEntries.size());\n        while (iterator.hasPrevious()) {\n            addEntryToBuffer(iterator.previous(), true);\n        }\n        nonBlockingFlush();\n    }\n\n    private void addEntryToBuffer(RequestEntryT entry, boolean insertAtHead) {\n        addEntryToBuffer(new RequestEntryWrapper<>(entry, getSizeInBytes(entry)), insertAtHead);\n    }\n\n    private void addEntryToBuffer(RequestEntryWrapper<RequestEntryT> entry, boolean insertAtHead) {\n        if (bufferedRequestEntries.isEmpty() && !existsActiveTimerCallback) {\n            registerCallback();\n        }\n\n        if (entry.getSize() > maxRecordSizeInBytes) {\n            throw new IllegalArgumentException(\n                    String.format(\n                            \"The request entry sent to the buffer was of size [%s], when the maxRecordSizeInBytes was set to [%s].\",\n                            entry.getSize(), maxRecordSizeInBytes));\n        }\n\n        bufferedRequestEntries.add(entry, insertAtHead);\n    }\n\n    /**\n     * In flight requests will be retried if the sink is still healthy. But if in-flight requests\n     * fail after a checkpoint has been triggered and Flink needs to recover from the checkpoint,\n     * the (failed) in-flight requests are gone and cannot be retried. Hence, there cannot be any\n     * outstanding in-flight requests when a commit is initialized.\n     *\n     * <p>To this end, all in-flight requests need to completed before proceeding with the commit.\n     */\n    @Override\n    public void flush(boolean flush) throws InterruptedException {","sourceCodeStart":398,"sourceCodeEnd":434,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java#L398-L434","documentation":"Thrown as an IllegalArgumentException by AsyncSinkWriter.addEntryToBuffer when a request entry's computed size (from getSizeInBytes) exceeds the configured maxRecordSizeInBytes limit. maxRecordSizeInBytes represents the maximum payload size the destination system accepts per record. This check prevents oversized records from entering the buffer where they would always fail on submission and could never be retried.","triggerScenarios":"An input element converts to a request entry whose getSizeInBytes() return value exceeds the maxRecordSizeInBytes configured via the sink builder's setMaxRecordSizeInBytes().","commonSituations":"A large record (e.g., a big JSON document or binary blob) exceeds the destination's per-record limit (e.g., AWS Kinesis 1MB, AWS Firehose limits); maxRecordSizeInBytes was left at a default that is too small for the data; getSizeInBytes returns an incorrect (inflated) size due to a bug in the ElementConverter.","solutions":["Pre-filter or split oversized records before they reach the sink (e.g., chunk large payloads into multiple smaller entries).","Increase maxRecordSizeInBytes in the sink builder if the destination supports larger records: builder.setMaxRecordSizeInBytes(largerValue).","Verify getSizeInBytes() in your AsyncSinkWriter returns the correct byte size — debug-log it for the failing record.","If the destination has a hard limit (e.g., Kinesis 1MB), implement a pre-processing step to split or compress records exceeding it."],"exampleFix":"// before — record exceeds max size\nbuilder.setMaxRecordSizeInBytes(1024 * 1024); // 1MB default\n// record is 2MB -> throws\n// after — split large records or increase limit\nbuilder.setMaxRecordSizeInBytes(4 * 1024 * 1024); // 4MB if destination supports it\n// Or split in ElementConverter:\n@Override\npublic String apply(String input, Context ctx) {\n    if (input.getBytes().length > MAX) { return split(input); }\n    return input;\n}","handlingStrategy":"validation","validationCode":"// Validate record size before adding to buffer\nlong recordSize = getSizeInBytes(requestEntry);\nif (recordSize > maxRecordSizeInBytes) {\n    // split, compress, or drop the record\n    throw new IllegalArgumentException(\"Record size \" + recordSize + \" exceeds max \" + maxRecordSizeInBytes);\n}","typeGuard":null,"tryCatchPattern":"try {\n    writer.invokeInternal(record);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"maxRecordSizeInBytes\")) {\n        // split or compress the record, or route to a dead-letter sink\n    }\n}","preventionTips":["Set maxRecordSizeInBytes to match the destination system's per-record limit.","Implement pre-splitting of large records in the ElementConverter.","Unit test getSizeInBytes() against edge-case records."],"tags":["async-sink","record-size","buffer","validation"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}