What causes "Could not get schema for configuration %s" in apache/beam?

Verify the table name in the SpannerChangeStreamConfig exactly matches a table covered by the change stream (names are compared as stored, and later lookups are case-insensitive).

apache/beam · error · RuntimeException

Could not get schema for configuration

Error message

Could not get schema for configuration %s

What it means

Thrown by getTableSchema in SpannerChangestreamsReadSchemaTransformProvider when the static TABLE_SCHEMAS map has no entry for the requested table after a mini-pipeline populates it via a DoFn. The cleanup code removes the entry; if it was never populated (schema capture failed or table name mismatch), the raw SpannerSchema object is null and this RuntimeException is raised.

Solutions

Verify the table name in the SpannerChangeStreamConfig exactly matches a table covered by the change stream (names are compared as stored, and later lookups are case-insensitive).
Check pipeline logs for failures in the schema-capturing mini pipeline (miniPipeline.run().waitUntilFinish()) — an upstream failure leaves TABLE_SCHEMAS empty.
Confirm the change stream actually tracks the requested table and that the metadata database/instance configuration is correct.
If running concurrent expansions, ensure they use distinct table keys or serialize access since TABLE_SCHEMAS entries are removed after first read.

Example fix

// before
config = SpannerChangeStreamReadSchemaTransformConfiguration.builder().setTable("Users").build(); // actual table is 'user'
// after
config = SpannerChangeStreamReadSchemaTransformConfiguration.builder().setTable("User").build();

Defensive patterns

Strategy: validation

Validate before calling

// Before expansion, ensure the config table matches the change stream's tracked table
Preconditions.checkArgument(
    trackedTables.stream().anyMatch(t -> t.equalsIgnoreCase(config.getTable())),
    "Table %s not covered by change stream", config.getTable());

Prevention

Assert the configured table equals the change stream's tracked table before expansion.
Do not share one provider expansion across concurrent threads; TABLE_SCHEMAS is a transient static map.
Watch for silent failures in the schema-capture mini pipeline in logs.

When it happens

Trigger: Calling tableChangesSchema/getTableSchema with a config whose table was never registered into TABLE_SCHEMAS by the schema-capturing mini pipeline; the pipeline failing silently or the DoFn not executing for that table; calling the expansion path twice concurrently so one consumer removes the entry before the other reads it.

Common situations: Misconfigured table name in the Spanner Read Change Stream config (case/name mismatch); change stream or database misconfiguration causing the schema lookup pipeline to produce no rows; races when the same provider instance is used by multiple concurrent expansions.

Understand the failure class

Background: Record Not Found Errors: "not found", RecordNotFound, and "was not found" — what they mean and how to fix them — this error's family across 28 libraries.

Related errors

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

Appendix: source

Thrown at sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:340

                            .withInstanceId(config.getInstanceId())
                            .withProjectId(config.getProjectId()),
                        sqlDialectView,
                        Sets.newHashSet(config.getTable())))
                .withSideInput("dialect", sqlDialectView))
        .apply(
            ParDo.of(
                new DoFn<SpannerSchema, String>() {
                  @ProcessElement
                  public void process(@DoFn.Element SpannerSchema schema) {
                    TABLE_SCHEMAS.put(config.getTable(), schema);
                  }
                }))
        .setCoder(StringUtf8Coder.of());
    miniPipeline.run().waitUntilFinish();
    // Clean up the static map from the object.
    SpannerSchema finalSchemaObj = TABLE_SCHEMAS.remove(config.getTable());
    if (finalSchemaObj == null) {
      throw new RuntimeException(
          String.format("Could not get schema for configuration %s", config));
    }
    return spannerSchemaToBeamSchema(finalSchemaObj, config.getTable());
  }

  private static Schema spannerSchemaToBeamSchema(
      SpannerSchema spannerSchema, final String tableName) {
    OptionalInt optionalIdx =
        IntStream.range(0, spannerSchema.getTables().size())
            .filter(idx -> spannerSchema.getTables().get(idx).equalsIgnoreCase(tableName))
            .findAny();
    if (!optionalIdx.isPresent()) {
      throw new IllegalArgumentException(
          String.format(
              "Unable to retrieve schema for table %s. Found only tables: [%s]",
              tableName, String.join(", ", spannerSchema.getTables())));
    }
    Schema.Builder schemaBuilder = Schema.builder();

View on GitHub (pinned to 12126d8942)