apache/beam · error · UnsupportedOperationException

maximumCacheSize is currently not supported for unbounded…

Error message

maximumCacheSize is currently not supported for unbounded streaming pipelines.

What it means

The Beam Iceberg connector can cap the number of cached table IDs processed using maximumCacheSize, implemented with Sample.any(), which requires bounded input. Unbounded (streaming) pipelines never finish, so sampling cannot cap the cache; the connector therefore rejects the combination at graph-construction time in expand().

Solutions

Remove withMaximumCacheSize(...) from the streaming pipeline configuration.
Run the pipeline in batch mode if cache capping is essential.
Use an alternative cache-eviction/time-based strategy supported for streaming (or upstream a streaming-safe cache size limit).

Example fix

// before
IcebergIO.write().to(table).withMaximumCacheSize(1000) // streaming pipeline
// after
IcebergIO.write().to(table) // drop maximumCacheSize for streaming

Defensive patterns

Strategy: validation

Validate before calling

if (isStreamingPipeline && options.getMaximumCacheSize() != null) {
  throw new IllegalArgumentException("Remove maximumCacheSize for streaming pipelines");
}

Type guard

Integer maxCacheSize = getMaximumCacheSize();
boolean streaming = pipeline.getOptions().as(StreamingOptions.class).isStreaming();
if (streaming) maxCacheSize = null; // null guards against the unsupported path

Try / catch

try {
  graph.apply(...);
} catch (UnsupportedOperationException e) {
  if (e.getMessage().contains("maximumCacheSize")) {
    // rebuild pipeline without maximumCacheSize
  } else { throw e; }
}

Prevention

Only set withMaximumCacheSize in batch pipelines.
Check StreamingOptions.isStreaming() before applying batch-only options.
Avoid copying batch pipeline config wholesale into streaming jobs.

When it happens

Trigger: Calling IcebergIO.writeDestinations (or similar) with .to(...).withMaximumCacheSize(n) applied to a streaming pipeline (isStreaming == true), causing expand() to throw UnsupportedOperationException during pipeline construction/translation.

Common situations: Reusing a batch-oriented cache-size tuning flag in a streaming job; copying a batch pipeline template and switching the runner to Flink/Spark/Dataflow streaming without removing the cache cap option.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

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

Appendix: source

Thrown at sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/TableMetadataDriver.java:279

    boolean isStreaming = input.isBounded() == PCollection.IsBounded.UNBOUNDED;

    PCollection<String> distinctTableIds;
    if (isStreaming) {
      Duration customInterval = getRefreshInterval();
      Duration interval =
          checkNotNull(customInterval != null ? customInterval : DEFAULT_REFRESH_INTERVAL);
      distinctTableIds =
          tableIds.apply(
              "DeduplicateTableIds", Deduplicate.<String>values().withDuration(interval));
    } else {
      distinctTableIds = tableIds.apply("DistinctTableIds", Distinct.create());
    }

    PCollection<String> cachedTableIds;
    Integer maxCacheSize = getMaximumCacheSize();
    if (maxCacheSize != null) {
      if (isStreaming) {
        throw new UnsupportedOperationException(
            "maximumCacheSize is currently not supported for unbounded streaming pipelines.");
      }
      cachedTableIds = distinctTableIds.apply("CapCacheSize", Sample.any(maxCacheSize));
    } else {
      cachedTableIds = distinctTableIds;
    }

    @Nullable Integer configuredBuckets = getPollingBuckets();
    int pollingBuckets = configuredBuckets != null ? configuredBuckets : DEFAULT_POLLING_BUCKETS;
    PCollection<String> pollingTableIds =
        cachedTableIds.apply(
            "ReshufflePollingBuckets",
            Reshuffle.<String>viaRandomKey().withNumBuckets(pollingBuckets));

    PCollection<KV<String, @Nullable SerializableTableSpec>> specs =
        pollingTableIds
            .apply("PollTableMetadata", ParDo.of(new CatalogPollingDoFn(getCatalogConfig())))
            .setCoder(

View on GitHub (pinned to 12126d8942)