{"record":{"id":"e80e1959297191f3","repo":"apache/seatunnel","slug":"all-candidate-sink-tables-were-skipped-in-flink-st","errorCode":null,"errorMessage":"All candidate sink tables were skipped in Flink starter.","messagePattern":"All candidate sink tables were skipped in Flink starter\\.","errorType":"exception","errorClass":"TaskExecuteException","httpStatus":null,"severity":"critical","filePath":"seatunnel-core/seatunnel-flink-starter/seatunnel-flink-13-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java","lineNumber":229,"sourceCode":"                                    .equalsIgnoreCase(envConfig.getString(\"job.mode\"));\n            DataStream<SeaTunnelRow> ds = stream.getDataStream();\n            if (isStreaming && sink instanceof SupportSchemaEvolutionSink) {\n                // insert broadcast-based schema operator to handle schema changes\n                ds =\n                        ds.transform(\n                                        \"BroadcastSchemaHandler\",\n                                        TypeInformation.of(SeaTunnelRow.class),\n                                        new BroadcastSchemaSinkOperator())\n                                .name(\"BroadcastSchemaHandler\")\n                                .setParallelism(parallelism);\n            }\n            DataStreamSink<SeaTunnelRow> dataStreamSink =\n                    ds.sinkTo(new FlinkSink<>(sink, stream.getCatalogTables(), parallelism))\n                            .name(String.format(\"%s-Sink\", sink.getPluginName()));\n            dataStreamSink.setParallelism(parallelism);\n        }\n        if (!createdAnySink && !skippedTables.isEmpty()) {\n            throw new TaskExecuteException(\n                    MultiTableFailureHelper.formatFailedTableSummary(\n                            \"All candidate sink tables were skipped in Flink starter.\",\n                            skippedTables));\n        }\n        if (createdAnySink && !skippedTables.isEmpty()) {\n            LOGGER.warn(\n                    MultiTableFailureHelper.formatFailedTableSummary(\n                            \"Some sink tables were skipped in Flink starter.\", skippedTables));\n        }\n        // the sink is the last stream\n        return null;\n    }\n\n    // if not support multi table, rollback\n    public SeaTunnelSink tryGenerateMultiTableSink(\n            Map<TablePath, SeaTunnelSink> sinks,\n            ReadonlyConfig sinkConfig,\n            ClassLoader classLoader) {","sourceCodeStart":211,"sourceCodeEnd":247,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-core/seatunnel-flink-starter/seatunnel-flink-13-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java#L211-L247","documentation":"TaskExecuteException thrown in SinkExecuteProcessor.execute when every candidate sink table failed pre-write checks and was skipped, so no DataStream sink was created at all. MultiTableFailureHelper formats a summary of the failed tables with their failure reasons so the user can see why all sinks were dropped.","triggerScenarios":"Running a Flink (1.13) starter job where each sink plugin (or each table in a multi-table sink) was skipped during execution — e.g. schema/catalog incompatibility, save-mode pre-check failure, or unsupported table — and skippedTables is non-empty while createdAnySink stays false.","commonSituations":"Sink connector rejecting the target table schema; database table missing and save-mode create disabled; connector not supporting the configured catalog table in Flink 1.3 starter.","solutions":["Read the per-table failure reasons in the formatted summary and fix the root cause (schema mismatch, missing table, etc.)","Enable/adjust save_mode config so the sink can create/repair the target table","Verify the sink connector supports the target table type on the Flink 1.13 starter"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// before submitting, verify target tables exist / are compatible\nfor (CatalogTable table : catalogTables) {\n    if (!sink.supportsTable(table)) {\n        log.warn(\"Table {} will be skipped\", table.getTablePath());\n    }\n}","typeGuard":null,"tryCatchPattern":"try {\n    processor.execute();\n} catch (TaskExecuteException e) {\n    log.error(\"All sink tables skipped: {}\", e.getMessage());\n    throw e;\n}","preventionTips":["Pre-check sink table schemas against source data before running the job","Enable save_mode so the sink can create missing tables","Test the connector against the target storage before production jobs"],"tags":["flink","sink","multi-table"],"backgroundTag":"schema-validation-failed","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}