apache/beam · error · NotImplementedException

Support for unbounded plugins is not implemented!

Error message

Support for unbounded plugins is not implemented!

What it means

CdapIO's expansion builds a bounded (batch) Hadoop-based read via the CDAP plugin wrapper. If the plugin is unbounded (streaming), this guard throws UnsupportedOperationException because CdapIO at this point only implements the bounded read path — streaming/unbounded CDAP plugins must use a different transform.

Solutions

Use a bounded/batch CDAP plugin for writes.
Convert the input PCollection to a bounded dataset before writing (e.g. window-triggered batch writes).
Wait for/implement SparkReceiverIO-based streaming write support (see the TODO in CdapIO).

Defensive patterns

Strategy: try-catch

Validate before calling

if (cdapPlugin.isUnbounded()) {
  throw new IllegalArgumentException("Streaming CDAP writes are unsupported; use a batch plugin");
}

Try / catch

try { transform.expand(input); } catch (NotImplementedException e) { /* switch to bounded plugin */ }

Prevention

Use batch CDAP plugins for write paths.
Check plugin boundedness before pipeline construction.
Track Beam releases for streaming write support.

When it happens

Trigger: Applying CdapIO.write() with a cdapPlugin whose isUnbounded() returns true, e.g. a streaming/stream-receiver source plugin being written to.

Common situations: Attempting to write to a CDAP sink from a streaming pipeline or using a streaming-oriented plugin config.

Related errors

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

Appendix: source

Thrown at sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/CdapIO.java:479

      PluginConfig pluginConfig = getPluginConfig();
      checkStateNotNull(pluginConfig, "withPluginConfig() is required");

      Class<K> keyClass = getKeyClass();
      checkStateNotNull(keyClass, "withKeyClass() is required");
      Class<V> valueClass = getValueClass();
      checkStateNotNull(valueClass, "withValueClass() is required");

      String locksDirPath = getLocksDirPath();
      checkStateNotNull(locksDirPath, "withLocksDirPath() is required");

      cdapPlugin
          .withConfig(pluginConfig)
          .withHadoopConfiguration(keyClass, valueClass)
          .prepareRun();

      if (cdapPlugin.isUnbounded()) {
        // TODO: implement SparkReceiverIO.<~>write()
        throw new NotImplementedException("Support for unbounded plugins is not implemented!");
      } else {
        Configuration hConf = cdapPlugin.getHadoopConfiguration();
        HadoopFormatIO.Write<K, V> writeHadoop;
        if (input.isBounded().equals(PCollection.IsBounded.UNBOUNDED)
            || !input.getWindowingStrategy().equals(WindowingStrategy.globalDefault())) {
          ConfigTransform<K, V> configTransform = new ConfigTransform<>(hConf);
          writeHadoop =
              HadoopFormatIO.<K, V>write()
                  .withConfigurationTransform(configTransform)
                  .withExternalSynchronization(new HDFSSynchronization(locksDirPath));
        } else {
          writeHadoop =
              HadoopFormatIO.<K, V>write()
                  .withConfiguration(hConf)
                  .withPartitioning()
                  .withExternalSynchronization(new HDFSSynchronization(locksDirPath));
        }
        return input.apply(writeHadoop);

View on GitHub (pinned to 12126d8942)