{"record":{"id":"d36b90eabbd80bf3","repo":"apache/seatunnel","slug":"interrupted-while-dispatching-postgresql-relation","errorCode":null,"errorMessage":"Interrupted while dispatching PostgreSQL relation change for ${table.id()}","messagePattern":"Interrupted while dispatching PostgreSQL relation change for (.+?)","errorType":"exception","errorClass":"DebeziumException","httpStatus":null,"severity":"warning","filePath":"seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java","lineNumber":371,"sourceCode":"        if (previousRelation != null\n                && RelationAwarePostgresSchema.hasSameRelationSchema(previousRelation, table)) {\n            return;\n        }\n        if (previousRelation == null && hasSameBaselineSchema(table)) {\n            return;\n        }\n\n        SourceRecord record =\n                PostgresRelationSchemaRecord.create(\n                        table,\n                        partition.getSourcePartition(),\n                        new HashMap<>(offsetContext.getOffset()),\n                        topicSelector.topicNameFor(table.id()));\n        try {\n            queue.enqueue(new DataChangeEvent(record));\n        } catch (InterruptedException e) {\n            Thread.currentThread().interrupt();\n            throw new DebeziumException(\n                    \"Interrupted while dispatching PostgreSQL relation change for \" + table.id(),\n                    e);\n        }\n    }\n\n    /** Compare the first RELATION for a table with initial or checkpoint-restored catalog state. */\n    private boolean hasSameBaselineSchema(Table relation) {\n        return relationSchemaBaseline.stream()\n                .filter(\n                        table ->\n                                Objects.equals(\n                                                table.getTablePath().getSchemaName(),\n                                                PostgresRelationSchemaChangeResolver\n                                                        .relationSchemaName(relation))\n                                        && Objects.equals(\n                                                table.getTablePath().getTableName(),\n                                                relation.id().table()))\n                .findFirst()","sourceCodeStart":353,"sourceCodeEnd":389,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java#L353-L389","documentation":"dispatchRelationSchemaChange enqueues a schema-change (RELATION) event onto the fetch task's change-event queue. If the enqueue is interrupted (thread interrupted while blocked on a full/closed queue), the thread's interrupt flag is restored and a DebeziumException 'Interrupted while dispatching...' is thrown, aborting the fetch task.","triggerScenarios":"The queue's enqueue blocks (downstream consumer stopped or slow, task being cancelled/shutdown) and the worker thread is interrupted during that wait.","commonSituations":"Job cancellation or failover interrupting reader threads while a DDL event is being dispatched; backpressure causing the queue to fill at the moment of shutdown; checkpoint/restore race conditions.","solutions":["Usually expected during job shutdown — restart/resubmit the job; the schema change will be re-read from the slot.","If it occurs under normal load, increase reader queue capacity or slow the upstream DDL churn.","Investigate what interrupted the thread (cancellation, timeout, failover) from surrounding logs.","Ensure checkpointing is enabled so the task can resume from the last committed LSN."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    fetchTask.run();\n} catch (DebeziumException e) {\n    if (e.getMessage().startsWith(\"Interrupted while dispatching\")) {\n        // expected during shutdown/cancel: resubmit or ignore\n        Thread.currentThread().interrupt();\n    } else {\n        throw e;\n    }\n}","preventionTips":["Avoid aggressive cancellation of jobs mid-DDL","Size reader queues to absorb schema-change bursts","Enable checkpointing for clean resume after interruption"],"tags":["postgresql","cdc","interruption","threading","schema-evolution"],"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"}