{"record":{"id":"d066130638e97be6","repo":"apache/beam","slug":"could-not-fetch-database-schema","errorCode":null,"errorMessage":"could not fetch database schema","messagePattern":"could not fetch database schema","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java","lineNumber":165,"sourceCode":"\n  @GetRestrictionCoder\n  public Coder<OffsetHolder> getRestrictionCoder() {\n    return SerializableCoder.of(OffsetHolder.class);\n  }\n\n  protected SourceRecord getOneRecord(Map<String, String> configuration) {\n    try {\n      SourceConnector connector = connectorClass.getDeclaredConstructor().newInstance();\n      connector.start(configuration);\n\n      SourceTask task = (SourceTask) connector.taskClass().getDeclaredConstructor().newInstance();\n      task.initialize(new BeamSourceTaskContext(null));\n      task.start(connector.taskConfigs(1).get(0));\n      List<SourceRecord> records = Lists.newArrayList();\n      int loops = 0;\n      while (records.size() == 0) {\n        if (loops > 3) {\n          throw new RuntimeException(\"could not fetch database schema\");\n        }\n        records = task.poll();\n        // Waiting for the Database snapshot to finish.\n        Thread.sleep(2000);\n        loops += 1;\n      }\n      task.stop();\n      connector.stop();\n      return records.get(0);\n    } catch (NoSuchMethodException\n        | InterruptedException\n        | InvocationTargetException\n        | IllegalAccessException\n        | InstantiationException e) {\n      throw new RuntimeException(\"Unexpected exception fetching database schema.\", e);\n    }\n  }\n","sourceCodeStart":147,"sourceCodeEnd":183,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java#L147-L183","documentation":"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.","triggerScenarios":"task.poll() returning empty lists for more than 3 consecutive iterations — connector misconfiguration, database unreachable, or the initial snapshot taking longer than ~6 seconds.","commonSituations":"Wrong connection credentials/host, large table snapshots exceeding the 2s-per-loop wait, database under load, network egress blocked from the worker.","solutions":["Verify connector config (hostname, port, user, password, table list) by running the connector outside Beam.","Increase the retry loop bound / sleep duration in KafkaSourceConsumerFn for large snapshots.","Check network connectivity from the Beam worker to the database (firewalls, VPC rules).","Confirm the database user has snapshot/replication privileges."],"exampleFix":"// before\nif (loops > 3) { throw new RuntimeException(\"could not fetch database schema\"); }\n// after\nif (loops > 30) { throw new RuntimeException(\"could not fetch database schema\"); }\nThread.sleep(5000);","handlingStrategy":"retry","validationCode":"// pre-flight: verify DB reachability before launching the pipeline\ntry (Connection c = DriverManager.getConnection(jdbcUrl, user, pass)) {\n  if (!c.isValid(5)) throw new IllegalStateException(\"DB unreachable\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  pipeline.run().waitUntilFinish();\n} catch (RuntimeException e) {\n  if (e.getMessage() != null && e.getMessage().equals(\"could not fetch database schema\")) {\n    // back off and retry the pipeline; snapshot may need more time\n  }\n}","preventionTips":["Validate connector config and credentials outside Beam first","Ensure network egress from workers to the database is open","Grant the connector user snapshot/replication privileges","Prefer incremental snapshots for very large tables"],"tags":["java","debezium","database","timeout","apache-beam"],"backgroundTag":"empty-result-set","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}