{"record":{"id":"ecbe68e7e64a5894","repo":"apache/beam","slug":"support-for-unbounded-plugins-is-not-implemented","errorCode":null,"errorMessage":"Support for unbounded plugins is not implemented!","messagePattern":"Support for unbounded plugins is not implemented!","errorType":"exception","errorClass":"NotImplementedException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/CdapIO.java","lineNumber":479,"sourceCode":"      PluginConfig pluginConfig = getPluginConfig();\n      checkStateNotNull(pluginConfig, \"withPluginConfig() is required\");\n\n      Class<K> keyClass = getKeyClass();\n      checkStateNotNull(keyClass, \"withKeyClass() is required\");\n      Class<V> valueClass = getValueClass();\n      checkStateNotNull(valueClass, \"withValueClass() is required\");\n\n      String locksDirPath = getLocksDirPath();\n      checkStateNotNull(locksDirPath, \"withLocksDirPath() is required\");\n\n      cdapPlugin\n          .withConfig(pluginConfig)\n          .withHadoopConfiguration(keyClass, valueClass)\n          .prepareRun();\n\n      if (cdapPlugin.isUnbounded()) {\n        // TODO: implement SparkReceiverIO.<~>write()\n        throw new NotImplementedException(\"Support for unbounded plugins is not implemented!\");\n      } else {\n        Configuration hConf = cdapPlugin.getHadoopConfiguration();\n        HadoopFormatIO.Write<K, V> writeHadoop;\n        if (input.isBounded().equals(PCollection.IsBounded.UNBOUNDED)\n            || !input.getWindowingStrategy().equals(WindowingStrategy.globalDefault())) {\n          ConfigTransform<K, V> configTransform = new ConfigTransform<>(hConf);\n          writeHadoop =\n              HadoopFormatIO.<K, V>write()\n                  .withConfigurationTransform(configTransform)\n                  .withExternalSynchronization(new HDFSSynchronization(locksDirPath));\n        } else {\n          writeHadoop =\n              HadoopFormatIO.<K, V>write()\n                  .withConfiguration(hConf)\n                  .withPartitioning()\n                  .withExternalSynchronization(new HDFSSynchronization(locksDirPath));\n        }\n        return input.apply(writeHadoop);","sourceCodeStart":461,"sourceCodeEnd":497,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/CdapIO.java#L461-L497","documentation":"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.","triggerScenarios":"Applying CdapIO.write() with a cdapPlugin whose isUnbounded() returns true, e.g. a streaming/stream-receiver source plugin being written to.","commonSituations":"Attempting to write to a CDAP sink from a streaming pipeline or using a streaming-oriented plugin config.","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)."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"if (cdapPlugin.isUnbounded()) {\n  throw new IllegalArgumentException(\"Streaming CDAP writes are unsupported; use a batch plugin\");\n}","typeGuard":null,"tryCatchPattern":"try { transform.expand(input); } catch (NotImplementedException e) { /* switch to bounded plugin */ }","preventionTips":["Use batch CDAP plugins for write paths.","Check plugin boundedness before pipeline construction.","Track Beam releases for streaming write support."],"tags":["java","apache-beam","cdap","streaming","unimplemented"],"backgroundTag":"method-not-implemented","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"}