{"record":{"id":"8dfdbea952c67fff","repo":"apache/seatunnel","slug":"bulk-response-error-8dfdbe","errorCode":"BULK_RESPONSE_ERROR","errorMessage":"bulk es error: ","messagePattern":"bulk es error: ","errorType":"error_code","errorClass":"ElasticsearchConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java","lineNumber":250,"sourceCode":"     * <p>The action is registered before multi-table resource injection but invoked after startup.\n     */\n    private void timerFlush() {\n        bulkEsWithRetry(this.esRestClient, this.requestEsList);\n    }\n\n    @Override\n    public void abortPrepare() {}\n\n    public synchronized void bulkEsWithRetry(\n            EsRestClient esRestClient, List<String> requestEsList) {\n        try {\n            RetryUtils.retryWithException(\n                    () -> {\n                        if (!requestEsList.isEmpty()) {\n                            String requestBody = String.join(\"\\n\", requestEsList) + \"\\n\";\n                            BulkResponse bulkResponse = esRestClient.bulk(requestBody);\n                            if (bulkResponse.isErrors()) {\n                                throw new ElasticsearchConnectorException(\n                                        ElasticsearchConnectorErrorCode.BULK_RESPONSE_ERROR,\n                                        \"bulk es error: \" + bulkResponse.getResponse());\n                            }\n                            return bulkResponse;\n                        }\n                        return null;\n                    },\n                    retryMaterial);\n            requestEsList.clear();\n        } catch (Exception e) {\n            throw new ElasticsearchConnectorException(\n                    CommonErrorCodeDeprecated.SQL_OPERATION_FAILED,\n                    \"ElasticSearch execute batch statement error\",\n                    e);\n        }\n    }\n\n    @Override","sourceCodeStart":232,"sourceCodeEnd":268,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java#L232-L268","documentation":"After sending a bulk request, ElasticsearchSinkWriter checks BulkResponse.isErrors(). If Elasticsearch reports per-item failures, it throws ElasticsearchConnectorException with BULK_RESPONSE_ERROR and the full response body. The batch is retried per the retry policy before failing.","triggerScenarios":"esRestClient.bulk(requestBody) returns a response with errors:true — e.g. mapping conflicts, index_not_found, version conflicts, rejected by cluster (429/oversized bulk), or invalid documents.","commonSituations":"Writing documents whose fields conflict with existing index mappings; bulk payloads exceeding http.max_content_length; index missing or closed; ES cluster under load rejecting bulk shards.","solutions":["Inspect the embedded response JSON for per-item error reasons (type/reason per item)","Fix mapping conflicts: delete/reindex the conflicting index or align the schema","Reduce batch size if the cluster rejected the bulk (429/oversized)","Ensure the target index exists and is open before writing","Check ES cluster health and disk watermarks (read-only indices block writes)"],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    writer.prepareCommit();\n} catch (ElasticsearchConnectorException e) {\n    if (e.getErrorCode() == ElasticsearchConnectorErrorCode.BULK_RESPONSE_ERROR) {\n        String body = e.getMessage(); // parse per-item errors and fix mappings/batch size\n        log.error(\"Bulk item failures: \" + body);\n    }\n    throw e;\n}","preventionTips":["Keep batch sizes under http.max_content_length and below cluster bulk limits","Verify index mappings match document fields before writing","Ensure target indices exist and are open with writable disk watermarks"],"tags":["elasticsearch","bulk-write","retry","cluster-response"],"backgroundTag":"http-error-response","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}