MSK Connector | MirrorMaker 2: Class not found

Question

I'm trying to replicate a topic in MSK using MirrorMaker 2, it's good to say as it is Serverless I don't have access to any instances to make sure that the jar is really present on the machine (classpath).

Basically, I want to completely change the topic name to XYZ, and I've been referring to a couple of people that tried by simply extending/implementing ReplicationPolicy. So I did it:
````
package mypackage.rep;

import org.apache.kafka.connect.mirror.ReplicationPolicy;

public class CustomReplicationPolicy implements ReplicationPolicy {

public String formatRemoteTopic(String sourceClusterAlias, String topic) {
        return "XYZ";
    }

public String topicSource(String topic) {
        return topic;
    }

public String upstreamTopic(String topic) {
        return null;
    }
}
````
Also referred it in the properties with:
````
replication.policy.class=mypackage.rep.CustomReplicationPolicy
````
But whenever I try to deploy it, it seems that the connector does not recognize it in the custom plugin zip:

`The connector configuration is invalid. Message: Connector configuration is invalid and contains the following 1 error(s): Invalid value mypackage.rep.CustomReplicationPolicy for configuration replication.policy.class: Class mypackage.rep.CustomReplicationPolicy could not be found.`

Output of jar tf connector.zip:
````
CustomReplicationPolicy-3.7.1.jar
__MACOSX/._CustomReplicationPolicy-3.7.1.jar
lib/
lib/CustomReplicationPolicy-3.7.1.jar
__MACOSX/lib/._CustomReplicationPolicy-3.7.1.jar
````

EdbE · Answer

Hi @Bombardi,
Your approach is great, but won't work specifically with managed Kafka Connect. When you submit a JAR as a plugin, it doesn't get added to the classpath of kafka connect worker. It goes into a plugin dir. Each plugin has it's own classloader for isolation of the dependencies, etc. This is how Kafka Connect workers are using plugins.
So, this means that the MirrorSource Connector won't *see* the libraries and classes for a custom replication policy.

Alternatively, if you are just trying to achieve same topic name replication, there is an IdentityReplicationPolicy available in Kafka 3. To use it, define your MM2 connector to run in MSK Connect using version 3.7

EDIT (addresses complete change of a topic name):

You can use SMT to change a topic name. Since the question explicitly provides one specific name to another specific name, the regex is concrete. Add this to the mirrorsource connector:

```
# other properties...
"transforms":"changeABCtoXYZ",
"transforms.changeABCtoXYZ.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.changeABCtoXYZ.regex": "ABC",
"transforms.changeABCtoXYZ.replacement": "XYZ",
# other properties...
```

re:Post Agent · Answer

The error you're encountering suggests that the custom replication policy class is not being found in the classpath of the MSK Connector. Here are some steps to troubleshoot and resolve this issue:

1. Ensure your custom JAR file is properly packaged:
   - The JAR file should contain your compiled class files in the correct package structure.
   - The manifest file in the JAR should correctly specify the main class if necessary.

2. Verify the contents of your connector.zip:
   - The CustomReplicationPolicy-3.7.1.jar should be directly in the root of the zip file, not just in the lib/ directory.
   - Remove any __MACOSX directories and hidden files, as they are not needed and may cause issues.

3. Double-check your connector configuration:
   - Make sure the replication.policy.class property is correctly set to the full class name, including the package.
   - Verify there are no typos in the class name or package name.

4. Consider using a custom plugin:
   - Instead of including your custom class directly in the connector configuration, you might need to create a custom plugin for MSK Connect.
   - This involves creating a manifest file that specifies your custom classes and packaging it with your JAR file.

5. Verify compatibility:
   - Ensure your custom class is compatible with the version of Kafka and MirrorMaker 2 used by MSK Connect.

6. Check MSK Connect logs:
   - If possible, review the MSK Connect logs for any additional error messages or stack traces that might provide more information about why the class isn't being found.

If you've verified all these points and are still encountering issues, you may need to contact AWS support for further assistance, as there might be limitations or specific requirements for custom classes in MSK Connect that are not immediately apparent.  
**Sources**  
[MSK connector failed to find class in MSK cluster with KRaft mode | AWS re:Post](https://repost.aws/questions/QUlAMJp4NSTTaSev2Dw5r46w/msk-connector-failed-to-find-class-in-msk-cluster-with-kraft-mode)  
[Connect to AWS MSK cluster from Local mac | AWS re:Post](https://repost.aws/questions/QUW0c-V5bSQYKB5iRNguIhiQ/connect-to-aws-msk-cluster-from-local-mac)

MSK Connector | MirrorMaker 2: Class not found

Relevant content