{"record":{"id":"3ca4927bd779337b","repo":"apache/seatunnel","slug":"buffer-add-failed","errorCode":"BUFFER_ADD_FAILED","errorMessage":"BatchBuffer is already closed.","messagePattern":"BatchBuffer is already closed\\.","errorType":"error_code","errorClass":"HugeGraphConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/buffer/BatchBuffer.java","lineNumber":125,"sourceCode":"            boolean checkVertex,\n            int maxInsertErrors,\n            String failureDataPath,\n            int subtaskIndex) {\n        // batchIntervalMs remains in the public signature for source compatibility. Timer flush is\n        // registered by HugeGraphSinkWriter with the engine instead of creating a connector thread.\n        this.batchSize = batchSize;\n        this.client = client;\n        this.batchFailureFallback = batchFailureFallback;\n        this.checkVertex = checkVertex;\n        this.maxInsertErrors = maxInsertErrors;\n        this.failureDataPath = failureDataPath;\n        this.subtaskIndex = subtaskIndex;\n        this.insertFailureCount = 0;\n    }\n\n    public synchronized void add(GraphElementEnvelope envelope) throws IOException {\n        if (closed) {\n            throw new HugeGraphConnectorException(\n                    HugeGraphConnectorErrorCode.BUFFER_ADD_FAILED,\n                    \"BatchBuffer is already closed.\");\n        }\n\n        try {\n            if (envelope.getElementType() == LabelType.VERTEX) {\n                vertexBuffer.add(envelope);\n                if (vertexBuffer.size() >= batchSize) {\n                    doFlushVertices();\n                }\n            } else {\n                edgeBuffer.add(envelope);\n                if (edgeBuffer.size() >= batchSize) {\n                    // Topology safety only matters when the server validates endpoints: with\n                    // check_vertex=true, flush pending vertices before the edges so no edge is sent\n                    // before its endpoints exist. With check_vertex=false the server already\n                    // accepts\n                    // orphan edges, so skip the forced (undersized) vertex flush and let the vertex","sourceCodeStart":107,"sourceCodeEnd":143,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-hugegraph/src/main/java/org/apache/seatunnel/connectors/seatunnel/hugegraph/buffer/BatchBuffer.java#L107-L143","documentation":"BatchBuffer.add is a synchronized method that appends a GraphElementEnvelope to the write buffer. If the buffer has already been closed (closed flag set during flushAndClose/close), any further add throws BUFFER_ADD_FAILED to prevent writing into a torn-down buffer.","triggerScenarios":"Calling add(envelope) after close() or after a flush triggered close (e.g., writer still processing records after checkpoint/prepareCommit closed the buffer; concurrent producer threads racing with close).","commonSituations":"Sink writer emitting late records after snapshot close; exception in flush path that closed the buffer but the caller retries adds; multi-threaded writes without coordinating with the sink lifecycle.","solutions":["Check BatchBuffer.isClosed() before calling add, or ensure all adds happen before close/flushAndClose.","Fix the writer lifecycle so records are drained before the sink closes the buffer.","If writing multi-threaded, close the buffer only after all producer threads have finished (join/barrier).","Inspect earlier logs for a flush exception that closed the buffer mid-stream and address the root cause."],"exampleFix":"// before\nbuffer.add(envelope);\n// after\nif (!buffer.isClosed()) {\n    buffer.add(envelope);\n} else {\n    // route to dead-letter or fail the batch cleanly\n}","handlingStrategy":"type-guard","validationCode":null,"typeGuard":"// java: gate adds on state\nif (buffer.isClosed()) { /* route elsewhere */ return; }\nbuffer.add(envelope);","tryCatchPattern":"try {\n    buffer.add(envelope);\n} catch (HugeGraphConnectorException e) {\n    if (e.getErrorCode() == HugeGraphConnectorErrorCode.BUFFER_ADD_FAILED) {\n        // buffer closed: stop producing, flush upstream, do not retry\n    } else throw e;\n}","preventionTips":["Close the buffer only after all writers finish (latch/join).","Never call add() after flushAndClose in finally blocks that may run during record processing.","Check earlier stack traces for the close trigger."],"tags":["hugegraph","buffer","lifecycle","race-condition"],"backgroundTag":"invalid-state-transition","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}