apache/beam · warning

Average partition bytes size has not been initialized…

Error message

Average partition bytes size has not been initialized, GetSize will always return 0, which will interfere with autoscaling.

What it means

DetectNewPartitionsDoFn.getSize, annotated with @GetSize, reports estimated work based on averagePartitionBytesSize. When the restriction tracker has not yet initialized this value (averagePartitionBytesSizeSet is false), the estimate is always 0, which breaks Beam Runner V2 autoscaling (the runner sees no work to scale on). The DoFn logs a warning each time GetSize is called in this state; the value returned is 0.

Solutions

1. Upgrade Apache Beam to a version that initializes averagePartitionBytesSize in DetectNewPartitionsDoFn's restriction tracker (the warning was addressed in later releases).
2. Verify the change stream and partition metadata tables are correctly configured so partition statistics are available.
3. If autoscaling matters, run on a runner/version where this warning does not appear, or manually fix the number of workers.
4. Check that the change stream name and start time are valid so partitions are detected and statistics populated.

Example fix

// before: stale Beam version with uninitialized size estimate
SpannerIO.readChangeStream()
    .withProjectId(projectId)
    .withInstanceId(instanceId)
    .withDatabaseId(databaseId)
    .withChangeStreamName(streamName);

// after: upgrade to a Beam release where DetectNewPartitionsDoFn
// initializes averagePartitionBytesSize before GetSize is queried

// At pipeline startup, log the Beam version and check for the fix in DetectNewPartitionsDoFn:
// grep -r "averagePartitionBytesSizeSet" sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/DetectNewPartitionsDoFn.java

When it happens

Trigger: Reading a Spanner change stream via SpannerIO.readChangeStream() on a runner/version where the average partition bytes size is never populated in the restriction tracker before GetSize is invoked — e.g. early in partition detection or when the metadata/dao has not supplied statistics yet.

Common situations: Running change-stream pipelines on Flink/Dataflow with autoscaling enabled and observing that the pipeline never scales because work is reported as 0; using a Beam version where DetectNewPartitionsDoFn does not initialize the size estimate from the partition metadata.

Related errors

• Both query and table cannot be specified at the same time…
• Cannot find Spanner table.
• Column has array of arrays which is prohibited in Spanner.
• Could not get schema for configuration
• Error processing struct to row

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/15be3286e8a1a120. 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/dofn/DetectNewPartitionsDoFn.java:127

  }

  /**
   * Uses an {@link TimestampRange} with a max range. This is because it does not know beforehand
   * how many partitions it will schedule.
   *
   * @return the timestamp range for the component
   */
  @GetInitialRestriction
  public TimestampRange initialRestriction(@Element PartitionMetadata partition) {
    final com.google.cloud.Timestamp createdAt = partition.getCreatedAt();
    return TimestampRange.of(
        TimestampUtils.previous(createdAt), com.google.cloud.Timestamp.MAX_VALUE);
  }

  @GetSize
  public double getSize(@Restriction TimestampRange restriction) {
    if (!averagePartitionBytesSizeSet) {
      LOG.warn(
          "Average partition bytes size has not been initialized, GetSize will always return 0, which will interfere with autoscaling.");
    }
    final com.google.cloud.Timestamp readTimestamp = restriction.getFrom();
    final PartitionMetadataDao dao = daoFactory.getPartitionMetadataDao();
    final long partitionsToSchedule = dao.countPartitionsCreatedAfter(readTimestamp);
    final long sizeEstimate = partitionsToSchedule * averagePartitionBytesSize;

    LOG.debug(
        "getSize() = {} ({} partitionsToSchedule * {} averagePartitionBytesSize)",
        sizeEstimate,
        partitionsToSchedule,
        averagePartitionBytesSize);
    return sizeEstimate;
  }

  @NewTracker
  public DetectNewPartitionsRangeTracker newTracker(@Restriction TimestampRange restriction) {
    return new DetectNewPartitionsRangeTracker(restriction);

View on GitHub (pinned to 12126d8942)