{"record":{"id":"3d58fb8e5af2fe37","repo":"apache/pulsar","slug":"sink-class-s-does-not-implement-the-correct-inter","errorCode":null,"errorMessage":"Sink class %s does not implement the correct interface","messagePattern":"Sink class (.+?) does not implement the correct interface","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java","lineNumber":226,"sourceCode":"            return resolveInterfaceTypeArguments(sourceClass, Source.class).get(0);\n        } else if (sourceClass.asErasure().isAssignableTo(BatchSource.class)) {\n            return resolveInterfaceTypeArguments(sourceClass, BatchSource.class).get(0);\n        } else {\n            throw new IllegalArgumentException(\n              String.format(\"Source class %s does not implement the correct interface\",\n                sourceClass.getActualName()));\n        }\n    }\n\n    public static TypeDefinition getSinkType(String className, TypePool typePool) {\n        return getSinkType(typePool.describe(className).resolve());\n    }\n\n    public static TypeDefinition getSinkType(TypeDefinition sinkClass) {\n        if (sinkClass.asErasure().isAssignableTo(Sink.class)) {\n            return resolveInterfaceTypeArguments(sinkClass, Sink.class).get(0);\n        } else {\n            throw new IllegalArgumentException(\n                    String.format(\"Sink class %s does not implement the correct interface\",\n                            sinkClass.getActualName()));\n        }\n    }\n\n    public static void downloadFromHttpUrl(String destPkgUrl, File targetFile) throws IOException {\n        final URL url = new URL(destPkgUrl);\n        final URLConnection connection = url.openConnection();\n        if (StringUtils.isNotEmpty(url.getUserInfo())) {\n            final AuthenticationDataBasic authBasic = new AuthenticationDataBasic(url.getUserInfo());\n            for (Map.Entry<String, String> header : authBasic.getHttpHeaders()) {\n                connection.setRequestProperty(header.getKey(), header.getValue());\n            }\n        }\n        try (InputStream in = connection.getInputStream()) {\n            log.info().attr(\"url\", destPkgUrl).attr(\"target\", targetFile.getAbsoluteFile())\n                    .log(\"Downloading function package\");\n            Files.copy(in, targetFile.toPath(), StandardCopyOption.REPLACE_EXISTING);","sourceCodeStart":208,"sourceCodeEnd":244,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java#L208-L244","documentation":"FunctionCommon.getSinkType inspects a user-provided Sink class to derive its input type via bytecode analysis. It requires the class to be assignable to org.apache.pulsar.functions.api.Sink; otherwise this IllegalArgumentException naming the offending class is thrown.","triggerScenarios":"Calling getSinkType(String className, TypePool) / getSinkType(TypeDefinition) with a class that does not implement Sink — e.g. passing a Source implementation as the sink class, a class implementing an old/renamed sink interface (pre java-function API), or a misconfigured className.","commonSituations":"Sink connector NAR built with the wrong entry class; migrating legacy pulsar-io sinks to the org.apache.pulsar.functions.api.Sink interface; typo in the sink class name in sinkConfig; shaded/relocated Sink interface in a fat jar breaking assignability.","solutions":["Make the class implement org.apache.pulsar.functions.api.Sink<T> with a concrete type argument","Correct the sinkConfig className so it references the actual Sink implementation","If migrating from the legacy io API, port the connector to org.apache.pulsar.functions.api.Sink","Verify the jar on the package URL contains the class with an unshaded/unrelocated Sink supertype"],"exampleFix":"// before\npublic class MySink implements Source<String> { ... } // registered as sink\n// after\npublic class MySink implements Sink<String> {\n  @Override public void open(Map<String,Object> config, SinkContext ctx) { }\n  @Override public void write(Record<String> record) { }\n  @Override public void close() { }\n}","handlingStrategy":"validation","validationCode":"// Verify interface before calling getSinkType\nClass<?> clazz = Class.forName(className);\nif (!org.apache.pulsar.functions.api.Sink.class.isAssignableFrom(clazz)) {\n  throw new IllegalArgumentException(className + \" must implement Sink\");\n}","typeGuard":"static boolean isValidSinkClass(String className) throws ClassNotFoundException {\n  return org.apache.pulsar.functions.api.Sink.class.isAssignableFrom(Class.forName(className));\n}","tryCatchPattern":"try {\n  TypeDefinition td = FunctionCommon.getSinkType(className, typePool);\n} catch (IllegalArgumentException e) {\n  log.error(\"Bad sink class: {}\", e.getMessage());\n  throw new InvalidFunctionDefinitionException(className + \" is not a Sink\");\n}","preventionTips":["Implement org.apache.pulsar.functions.api.Sink<T> with a concrete type argument","Verify sinkConfig className points at the Sink implementation, not the source","Avoid relocating/shading the pulsar-functions-api package in connector fat jars","Compile-check the connector module against the pulsar-functions-api dependency"],"tags":["pulsar-functions","sink","interface","configuration"],"backgroundTag":"class-does-not-implement-interface","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}