apache/seatunnel · error · RabbitmqConnectorException

API-01

API-01

Error message

Cannot find schema or tableId for queue: %s. This queue is not configured in tables_configs. Available queues: %s

What it means

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.

Solutions

  1. Ensure every queue the consumer actually receives messages from is listed in tables-configs with a valid schema
  2. Fix queue-name spelling so it matches the actual broker queue and the splitId in messages
  3. Restart the job after config changes so splits and schemaMap are rebuilt consistently

Example fix

// before
tables-configs = [
  { queue-name = "order", schema = { fields { id = int } } } // typo: broker queue is "orders"
]
// after
tables-configs = [
  { queue-name = "orders", schema = { fields { id = int } } }
]
Defensive patterns

Strategy: validation

Validate before calling

// before submitting, ensure every consumed queue is declared in tables-configs
java.util.Set<String> configured = tablesConfigs.stream()
    .map(t -> (String) t.get("queue-name")).collect(java.util.stream.Collectors.toSet());
if (!configured.containsAll(consumedQueueNames)) {
  throw new IllegalArgumentException("Queues missing from tables-configs: " + consumedQueueNames);
}

Try / catch

try {
  reader.pollNext();
} catch (RabbitmqConnectorException e) {
  if (e.getMessage() != null && e.getMessage().contains("Cannot find schema or tableId for queue")) {
    System.err.println("Add the missing queue (with schema) to tables-configs and restart the job");
  } else {
    throw e;
  }
}

Prevention

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/3858aefe1ddc7e69. Report an issue: GitHub.

Appendix: source

Thrown at seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/source/RabbitmqSourceReader.java:186

                if (schema != null && exactTableId != null) {
                    SeaTunnelRow row = schema.deserialize(delivery.getBody());

                    if (row != null) {
                        // Tag the row with its specific Table ID to ensure downstream sinks route
                        // it correctly
                        row.setTableId(exactTableId);
                        output.collect(row);
                    }
                } else {
                    String errorMsg =
                            String.format(
                                    "Cannot find schema or tableId for queue: %s. "
                                            + "This queue is not configured in tables_configs. "
                                            + "Available queues: %s",
                                    message.getSplitId(), schemaMap.keySet());
                    log.error(errorMsg);
                    throw new RabbitmqConnectorException(
                            SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED, errorMsg);
                }
            }
        }

        // Bounded mode logic: Stop the job if all splits have been consumed and the queue is empty
        if (Boundedness.BOUNDED.equals(context.getBoundedness()) && noMoreSplitsAssigned) {
            if (message == null && queue.isEmpty()) {
                log.info(
                        "No more splits assigned, queue is empty, and polling timed out. Signaling end of input.");
                context.signalNoMoreElement();
            }
        }
    }

    @Override
    public void addSplits(List<RabbitmqSplit> splits) {
        // Dynamically start consuming from newly assigned queues (splits)

View on GitHub (pinned to cf67b549a7)