{"record":{"id":"3858aefe1ddc7e69","repo":"apache/seatunnel","slug":"api-01-3858ae","errorCode":"API-01","errorMessage":"Cannot find schema or tableId for queue: %s. This queue is not configured in tables_configs. Available queues: %s","messagePattern":"Cannot find schema or tableId for queue: (.+?)\\. This queue is not configured in tables_configs\\. Available queues: (.+?)","errorType":"error_code","errorClass":"RabbitmqConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceReader.java","lineNumber":186,"sourceCode":"\n                if (schema != null && exactTableId != null) {\n                    SeaTunnelRow row = schema.deserialize(delivery.getBody());\n\n                    if (row != null) {\n                        // Tag the row with its specific Table ID to ensure downstream sinks route\n                        // it correctly\n                        row.setTableId(exactTableId);\n                        output.collect(row);\n                    }\n                } else {\n                    String errorMsg =\n                            String.format(\n                                    \"Cannot find schema or tableId for queue: %s. \"\n                                            + \"This queue is not configured in tables_configs. \"\n                                            + \"Available queues: %s\",\n                                    message.getSplitId(), schemaMap.keySet());\n                    log.error(errorMsg);\n                    throw new RabbitmqConnectorException(\n                            SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED, errorMsg);\n                }\n            }\n        }\n\n        // Bounded mode logic: Stop the job if all splits have been consumed and the queue is empty\n        if (Boundedness.BOUNDED.equals(context.getBoundedness()) && noMoreSplitsAssigned) {\n            if (message == null && queue.isEmpty()) {\n                log.info(\n                        \"No more splits assigned, queue is empty, and polling timed out. Signaling end of input.\");\n                context.signalNoMoreElement();\n            }\n        }\n    }\n\n    @Override\n    public void addSplits(List<RabbitmqSplit> splits) {\n        // Dynamically start consuming from newly assigned queues (splits)","sourceCodeStart":168,"sourceCodeEnd":204,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceReader.java#L168-L204","documentation":"RabbitmqSourceReader.pollNext throws RabbitmqConnectorException(CONFIG_VALIDATION_FAILED) when a message arrives for a queue (splitId) that has no entry in schemaMap — i.e. the queue is not declared in tables-configs, so no schema or tableId can be resolved for it.","triggerScenarios":"At runtime a message is consumed whose split/queue id is absent from the schemaMap built from tables-configs — e.g. consuming messages from queues beyond the configured set, or schemaMap built before tables-configs was parsed.","commonSituations":"RabbitMQ sends messages for queues the job didn't declare; typo in a queue-name inside tables-configs so its schema never lands in schemaMap; config edited/renamed queue while reader was running with old splits.","solutions":["Ensure every queue the consumer actually receives messages from is listed in tables-configs with a valid schema","Fix queue-name spelling so it matches the actual broker queue and the splitId in messages","Restart the job after config changes so splits and schemaMap are rebuilt consistently"],"exampleFix":"// before\ntables-configs = [\n  { queue-name = \"order\", schema = { fields { id = int } } } // typo: broker queue is \"orders\"\n]\n// after\ntables-configs = [\n  { queue-name = \"orders\", schema = { fields { id = int } } }\n]","handlingStrategy":"validation","validationCode":"// before submitting, ensure every consumed queue is declared in tables-configs\njava.util.Set<String> configured = tablesConfigs.stream()\n    .map(t -> (String) t.get(\"queue-name\")).collect(java.util.stream.Collectors.toSet());\nif (!configured.containsAll(consumedQueueNames)) {\n  throw new IllegalArgumentException(\"Queues missing from tables-configs: \" + consumedQueueNames);\n}","typeGuard":null,"tryCatchPattern":"try {\n  reader.pollNext();\n} catch (RabbitmqConnectorException e) {\n  if (e.getMessage() != null && e.getMessage().contains(\"Cannot find schema or tableId for queue\")) {\n    System.err.println(\"Add the missing queue (with schema) to tables-configs and restart the job\");\n  } else {\n    throw e;\n  }\n}","preventionTips":["List every queue the binding/consumer can receive from in tables-configs","Avoid renaming broker queues while a job is running","Verify schemaMap contents (log available queues) during development"],"tags":["rabbitmq","runtime","schema","tables-configs"],"backgroundTag":"resource-not-found","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"}