{"record":{"id":"00cc57102b4bc2bc","repo":"apache/beam","slug":"provided-class-should-be-source-or-sink-plugin","errorCode":null,"errorMessage":"Provided class should be source or sink plugin","messagePattern":"Provided class should be source or sink plugin","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/Plugin.java","lineNumber":206,"sourceCode":"  }\n\n  /** Gets a Hadoop type. */\n  private PluginConstants.Hadoop getHadoopType() {\n    return getPluginType() == PluginConstants.PluginType.SOURCE\n        ? PluginConstants.Hadoop.SOURCE\n        : PluginConstants.Hadoop.SINK;\n  }\n\n  /** Gets value of a plugin type. */\n  public static PluginConstants.PluginType initPluginType(Class<?> pluginClass)\n      throws IllegalArgumentException {\n    if (StreamingSource.class.isAssignableFrom(pluginClass)\n        || BatchSource.class.isAssignableFrom(pluginClass)) {\n      return PluginConstants.PluginType.SOURCE;\n    } else if (BatchSink.class.isAssignableFrom(pluginClass)) {\n      return PluginConstants.PluginType.SINK;\n    } else {\n      throw new IllegalArgumentException(\"Provided class should be source or sink plugin\");\n    }\n  }\n\n  /** Initializes {@link BatchContextImpl} for CDAP plugin. */\n  public static BatchContextImpl initContext(Class<?> cdapPluginClass) {\n    // Init context and determine input or output\n    Class<?> contextClass;\n    List<Method> methods = new ArrayList<>(Arrays.asList(cdapPluginClass.getDeclaredMethods()));\n    Class<?> cdapPluginSuperclass = cdapPluginClass.getSuperclass();\n    if (cdapPluginSuperclass != null) {\n      methods.addAll(Arrays.asList(cdapPluginSuperclass.getDeclaredMethods()));\n    }\n    for (Method method : methods) {\n      if (method.getName().equals(PREPARE_RUN_METHOD_NAME)) {\n        contextClass = method.getParameterTypes()[0];\n        if (contextClass.equals(BatchSourceContext.class)) {\n          return new BatchSourceContextImpl();\n        } else if (contextClass.equals(BatchSinkContext.class)) {","sourceCodeStart":188,"sourceCodeEnd":224,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/Plugin.java#L188-L224","documentation":"Plugin.initPluginType inspects the CDAP plugin class to classify it as a SOURCE or SINK by checking assignability from StreamingSource/BatchSource/BatchSink. If the provided class implements none of these CDAP interfaces, initPluginType cannot classify it and throws IllegalArgumentException. This guard exists because the Beam CDAP adapter can only wrap plugins that implement one of the recognized source/sink interfaces.","triggerScenarios":"Calling Plugin.of(...) or any code path that invokes initPluginType with a Class<?> that does not extend BatchSource, StreamingSource, or BatchSink (e.g., a plugin that implements BatchAggregator, BatchRunnable, or a custom interface, or a wrong class passed by mistake such as the config class instead of the plugin class).","commonSituations":"Passing a CDAP transform/aggregator plugin where a source or sink is required; typos in generics causing the config class or plugin config to be handed in; upgrading a plugin that renamed its base class so it no longer extends BatchSource/BatchSink; building a plugin against a different CDAP API version where the interfaces live in another package.","solutions":["Verify the class passed to the CDAP IO source/sink extends io.cdap.plugin BatchSource, StreamingSource, or BatchSink","If you intended a transform/aggregator plugin, use the appropriate CDAP transform API instead of the Beam CDAP source/sink wrapper","Check that your plugin depends on the CDAP API artifacts providing BatchSource/BatchSink so isAssignableFrom actually matches","Print/log pluginClass.getInterfaces() and superclasses to confirm the hierarchy at runtime"],"exampleFix":"// before\nPipeline.apply(CdapSourceIO.of(MyAggregator.class, config, typeDescriptor));\n// after\nPipeline.apply(CdapSourceIO.of(MyBatchSource.class /* extends BatchSource */, config, typeDescriptor));","handlingStrategy":"validation","validationCode":"if (!BatchSource.class.isAssignableFrom(pluginClass)\n    && !StreamingSource.class.isAssignableFrom(pluginClass)\n    && !BatchSink.class.isAssignableFrom(pluginClass)) {\n  throw new IllegalArgumentException(pluginClass + \" is not a CDAP source/sink plugin\");\n}","typeGuard":"boolean isCdapSourceOrSink(Class<?> c) {\n  return BatchSource.class.isAssignableFrom(c)\n      || StreamingSource.class.isAssignableFrom(c)\n      || BatchSink.class.isAssignableFrom(c);\n}","tryCatchPattern":"try {\n  CdapSourceIO.of(pluginClass, config, typeDescriptor);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().contains(\"source or sink plugin\")) { /* fall back to correct plugin class */ }\n  throw e;\n}","preventionTips":["Always pass the class that extends BatchSource/StreamingSource/BatchSink, never the config class","Add a unit test asserting isCdapSourceOrSink(MyPlugin.class) for each plugin you wrap","Check the plugin's superclass hierarchy after CDAP version upgrades"],"tags":["java","cdap","beam","illegal-argument","plugin-type"],"backgroundTag":"invalid-argument-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}