{"record":{"id":"2eda437e2bf573d6","repo":"apache/beam","slug":"maximumcachesize-is-currently-not-supported-for-unbounded","errorCode":null,"errorMessage":"maximumCacheSize is currently not supported for unbounded streaming pipelines.","messagePattern":"maximumCacheSize is currently not supported for unbounded streaming pipelines\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/TableMetadataDriver.java","lineNumber":279,"sourceCode":"    boolean isStreaming = input.isBounded() == PCollection.IsBounded.UNBOUNDED;\n\n    PCollection<String> distinctTableIds;\n    if (isStreaming) {\n      Duration customInterval = getRefreshInterval();\n      Duration interval =\n          checkNotNull(customInterval != null ? customInterval : DEFAULT_REFRESH_INTERVAL);\n      distinctTableIds =\n          tableIds.apply(\n              \"DeduplicateTableIds\", Deduplicate.<String>values().withDuration(interval));\n    } else {\n      distinctTableIds = tableIds.apply(\"DistinctTableIds\", Distinct.create());\n    }\n\n    PCollection<String> cachedTableIds;\n    Integer maxCacheSize = getMaximumCacheSize();\n    if (maxCacheSize != null) {\n      if (isStreaming) {\n        throw new UnsupportedOperationException(\n            \"maximumCacheSize is currently not supported for unbounded streaming pipelines.\");\n      }\n      cachedTableIds = distinctTableIds.apply(\"CapCacheSize\", Sample.any(maxCacheSize));\n    } else {\n      cachedTableIds = distinctTableIds;\n    }\n\n    @Nullable Integer configuredBuckets = getPollingBuckets();\n    int pollingBuckets = configuredBuckets != null ? configuredBuckets : DEFAULT_POLLING_BUCKETS;\n    PCollection<String> pollingTableIds =\n        cachedTableIds.apply(\n            \"ReshufflePollingBuckets\",\n            Reshuffle.<String>viaRandomKey().withNumBuckets(pollingBuckets));\n\n    PCollection<KV<String, @Nullable SerializableTableSpec>> specs =\n        pollingTableIds\n            .apply(\"PollTableMetadata\", ParDo.of(new CatalogPollingDoFn(getCatalogConfig())))\n            .setCoder(","sourceCodeStart":261,"sourceCodeEnd":297,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/TableMetadataDriver.java#L261-L297","documentation":"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().","triggerScenarios":"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.","commonSituations":"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.","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)."],"exampleFix":"// before\nIcebergIO.write().to(table).withMaximumCacheSize(1000) // streaming pipeline\n// after\nIcebergIO.write().to(table) // drop maximumCacheSize for streaming","handlingStrategy":"validation","validationCode":"if (isStreamingPipeline && options.getMaximumCacheSize() != null) {\n  throw new IllegalArgumentException(\"Remove maximumCacheSize for streaming pipelines\");\n}","typeGuard":"Integer maxCacheSize = getMaximumCacheSize();\nboolean streaming = pipeline.getOptions().as(StreamingOptions.class).isStreaming();\nif (streaming) maxCacheSize = null; // null guards against the unsupported path","tryCatchPattern":"try {\n  graph.apply(...);\n} catch (UnsupportedOperationException e) {\n  if (e.getMessage().contains(\"maximumCacheSize\")) {\n    // rebuild pipeline without maximumCacheSize\n  } else { throw e; }\n}","preventionTips":["Only set withMaximumCacheSize in batch pipelines.","Check StreamingOptions.isStreaming() before applying batch-only options.","Avoid copying batch pipeline config wholesale into streaming jobs."],"tags":["java","beam","streaming","iceberg","unsupported-option"],"backgroundTag":"unsupported-operation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}