apache/beam · error · IllegalArgumentException

Provided class should be source or sink plugin

Error message

Provided class should be source or sink plugin

What it means

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.

Source

Thrown at sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/Plugin.java:206

  }

  /** Gets a Hadoop type. */
  private PluginConstants.Hadoop getHadoopType() {
    return getPluginType() == PluginConstants.PluginType.SOURCE
        ? PluginConstants.Hadoop.SOURCE
        : PluginConstants.Hadoop.SINK;
  }

  /** Gets value of a plugin type. */
  public static PluginConstants.PluginType initPluginType(Class<?> pluginClass)
      throws IllegalArgumentException {
    if (StreamingSource.class.isAssignableFrom(pluginClass)
        || BatchSource.class.isAssignableFrom(pluginClass)) {
      return PluginConstants.PluginType.SOURCE;
    } else if (BatchSink.class.isAssignableFrom(pluginClass)) {
      return PluginConstants.PluginType.SINK;
    } else {
      throw new IllegalArgumentException("Provided class should be source or sink plugin");
    }
  }

  /** Initializes {@link BatchContextImpl} for CDAP plugin. */
  public static BatchContextImpl initContext(Class<?> cdapPluginClass) {
    // Init context and determine input or output
    Class<?> contextClass;
    List<Method> methods = new ArrayList<>(Arrays.asList(cdapPluginClass.getDeclaredMethods()));
    Class<?> cdapPluginSuperclass = cdapPluginClass.getSuperclass();
    if (cdapPluginSuperclass != null) {
      methods.addAll(Arrays.asList(cdapPluginSuperclass.getDeclaredMethods()));
    }
    for (Method method : methods) {
      if (method.getName().equals(PREPARE_RUN_METHOD_NAME)) {
        contextClass = method.getParameterTypes()[0];
        if (contextClass.equals(BatchSourceContext.class)) {
          return new BatchSourceContextImpl();
        } else if (contextClass.equals(BatchSinkContext.class)) {

View on GitHub (pinned to 12126d8942)

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

Example fix

// before
Pipeline.apply(CdapSourceIO.of(MyAggregator.class, config, typeDescriptor));
// after
Pipeline.apply(CdapSourceIO.of(MyBatchSource.class /* extends BatchSource */, config, typeDescriptor));

Defensive patterns

Strategy: validation

Validate before calling

if (!BatchSource.class.isAssignableFrom(pluginClass)
    && !StreamingSource.class.isAssignableFrom(pluginClass)
    && !BatchSink.class.isAssignableFrom(pluginClass)) {
  throw new IllegalArgumentException(pluginClass + " is not a CDAP source/sink plugin");
}

Type guard

boolean isCdapSourceOrSink(Class<?> c) {
  return BatchSource.class.isAssignableFrom(c)
      || StreamingSource.class.isAssignableFrom(c)
      || BatchSink.class.isAssignableFrom(c);
}

Try / catch

try {
  CdapSourceIO.of(pluginClass, config, typeDescriptor);
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("source or sink plugin")) { /* fall back to correct plugin class */ }
  throw e;
}

Prevention

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

When it happens

Trigger: 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).

Common situations: 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.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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