{"record":{"id":"ba569ee52f16e3ea","repo":"apache/seatunnel","slug":"stage-load-failed","errorCode":"STAGE_LOAD_FAILED","errorMessage":"offer data to queue timeout, exceed ","messagePattern":"offer data to queue timeout, exceed ","errorType":"error_code","errorClass":"SelectDBConnectorException","httpStatus":null,"severity":"critical","filePath":"seatunnel-connectors-v2/connector-selectdb-cloud/src/main/java/org/apache/seatunnel/connectors/selectdb/sink/writer/SelectDBStageLoad.java","lineNumber":154,"sourceCode":"        if (buffer == null) {\n            return;\n        }\n        String fileName =\n                labelGenerator.generateLabel(currentCheckpointID, fileNum.getAndIncrement());\n        buffer.setFileName(fileName);\n        RecordBuffer tmpBuff = buffer;\n        log.info(\"flush buffer to queue, actual queue size {}\", queue.size());\n        offer(tmpBuff);\n        if (waitUtilDone) {\n            waitAsyncLoadFinish();\n        }\n        this.buffer = new RecordBuffer(this.lineDelimiter);\n    }\n\n    private void offer(RecordBuffer buffer) throws InterruptedException {\n        checkFlushException();\n        if (!queue.offer(buffer, 600 * 1000, TimeUnit.MILLISECONDS)) {\n            throw new SelectDBConnectorException(\n                    SelectDBConnectorErrorCode.STAGE_LOAD_FAILED,\n                    \"offer data to queue timeout, exceed \");\n        }\n    }\n\n    private void checkFlushException() {\n        if (exception.get() != null) {\n            throw new SelectDBConnectorException(\n                    SelectDBConnectorErrorCode.STAGE_LOAD_FAILED, exception.get());\n        }\n    }\n\n    private void waitAsyncLoadFinish() throws InterruptedException {\n        for (int i = 0; i < selectdbConfig.getFlushQueueSize() + 1; i++) {\n            offer(new RecordBuffer());\n        }\n    }\n","sourceCodeStart":136,"sourceCodeEnd":172,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-selectdb-cloud/src/main/java/org/apache/seatunnel/connectors/selectdb/sink/writer/SelectDBStageLoad.java#L136-L172","documentation":"SelectDBStageLoad feeds RecordBuffers into an internal ArrayBlockingQueue consumed by the async stage-load thread. offer() waits up to 600 seconds for space; if the queue is still full (consumer stalled or dead), it throws SelectDBConnectorException with code STAGE_LOAD_FAILED.","triggerScenarios":"The async load thread is blocked or crashed (e.g. slow/hung HTTP upload to SelectDB, or it died with an unhandled exception) so the queue stays full for 600s while flush() or waitAsyncLoadFinish() calls offer().","commonSituations":"SelectDB Cloud endpoint slow or unreachable (network issues, DNS, proxy); load thread killed earlier by a prior exception; batch too large making uploads take longer than the offer window in a saturated pipeline.","solutions":["Check worker logs for an earlier exception in the async load thread — the queue-full timeout is usually a secondary symptom","Verify network connectivity and response times to the SelectDB Cloud internal stage endpoint","Reduce sink batch size/buffer limits to lower per-upload latency and increase throughput","Restart the job after fixing the root cause; consider increasing the 600s offer timeout for very slow storage"],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// health-check the SelectDB endpoint and monitor queue depth before/at flush\nif (queue.remainingCapacity() == 0) { log.warn(\"stage-load queue saturated\"); }","typeGuard":null,"tryCatchPattern":"try { sink.flush(); } catch (SelectDBConnectorException e) { if (e.getErrorCode() == SelectDBConnectorErrorCode.STAGE_LOAD_FAILED) { /* restart job after checking load-thread logs */ } }","preventionTips":["Monitor network latency to the SelectDB Cloud endpoint","Reduce batch/buffer sizes to keep uploads short","Check worker logs for earlier async-load-thread exceptions","Alert on sustained queue saturation"],"tags":["timeout","queue","selectdb"],"backgroundTag":"request-timeout","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"}