apache/beam · error · RuntimeException

could not fetch database schema

Error message

could not fetch database schema

What it means

getOneRecord starts a Debezium task briefly to poll one record and learn the database schema. If after 4 loops (3 retries × ~2s sleeps) no records arrive, it assumes the snapshot/schema fetch failed and throws. The error means the connector produced nothing within the timeout window.

Solutions

1. Verify connector config (hostname, port, user, password, table list) by running the connector outside Beam.
2. Increase the retry loop bound / sleep duration in KafkaSourceConsumerFn for large snapshots.
3. Check network connectivity from the Beam worker to the database (firewalls, VPC rules).
4. Confirm the database user has snapshot/replication privileges.

Example fix

// before
if (loops > 3) { throw new RuntimeException("could not fetch database schema"); }
// after
if (loops > 30) { throw new RuntimeException("could not fetch database schema"); }
Thread.sleep(5000);

// pre-flight: verify DB reachability before launching the pipeline
try (Connection c = DriverManager.getConnection(jdbcUrl, user, pass)) {
  if (!c.isValid(5)) throw new IllegalStateException("DB unreachable");
}

try {
  pipeline.run().waitUntilFinish();
} catch (RuntimeException e) {
  if (e.getMessage() != null && e.getMessage().equals("could not fetch database schema")) {
    // back off and retry the pipeline; snapshot may need more time
  }
}

When it happens

Trigger: task.poll() returning empty lists for more than 3 consecutive iterations — connector misconfiguration, database unreachable, or the initial snapshot taking longer than ~6 seconds.

Common situations: Wrong connection credentials/host, large table snapshots exceeding the 2s-per-loop wait, database under load, network egress blocked from the worker.

Understand the failure class

Background: EmptyResultError / "no results found": when an API or scraper succeeds but returns zero rows — this error's family across 9 libraries.

Related errors

• Error occurred when consuming changes from Database.
• Cannot create enum from value!
• host must not be empty.
• Invalid Debezium connection property ''. Expected key=value…
• port must be between 1 and 65535, but was .

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/d066130638e97be6. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java:165

  @GetRestrictionCoder
  public Coder<OffsetHolder> getRestrictionCoder() {
    return SerializableCoder.of(OffsetHolder.class);
  }

  protected SourceRecord getOneRecord(Map<String, String> configuration) {
    try {
      SourceConnector connector = connectorClass.getDeclaredConstructor().newInstance();
      connector.start(configuration);

      SourceTask task = (SourceTask) connector.taskClass().getDeclaredConstructor().newInstance();
      task.initialize(new BeamSourceTaskContext(null));
      task.start(connector.taskConfigs(1).get(0));
      List<SourceRecord> records = Lists.newArrayList();
      int loops = 0;
      while (records.size() == 0) {
        if (loops > 3) {
          throw new RuntimeException("could not fetch database schema");
        }
        records = task.poll();
        // Waiting for the Database snapshot to finish.
        Thread.sleep(2000);
        loops += 1;
      }
      task.stop();
      connector.stop();
      return records.get(0);
    } catch (NoSuchMethodException
        | InterruptedException
        | InvocationTargetException
        | IllegalAccessException
        | InstantiationException e) {
      throw new RuntimeException("Unexpected exception fetching database schema.", e);
    }
  }

View on GitHub (pinned to 12126d8942)