Custom replication policy for MirrorSourceConnector

Kafka Connect Self-Managed Connectors
1

Hi Guys,

we have an already running Kakfa Connector Cluster.

I want to add a MirrorSourceConnector with a custom replication policy:

curl --request POST \
--url http://localhost:9083/connectors \
--header 'Accept: application/json' \
--header 'Content-Type: application/json' \
--data '{
"name" : "cluster-source",
"config":{
"connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
...

"replication.policy.class" : "org.example.CustomReplicationPolicy",
...
}
}'

This leads to an error:

{

"error_code": 400,

"message": "Connector configuration is invalid and contains the following 1 error(s):\nInvalid value org.example.CustomReplicationPolicy for configuration replication.policy.class: Class org.example.CustomReplicationPolicy could not be found.\nYou can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`"

}

I added the jar which contains CustomReplicationPolicy to the /lib path but it won’t work.

How can I use my CustomReplicationPolicy with the MirrorSourceConnector?

I’m using version 2.8.

Setting up the MirrorMaker with /bin/connect-mirror-maker.sh and the jar file in /lib works, but unfortunately I have to configure the mirror maker connectors one by one in our enviroment.

BR
Toni

3

I found the issue. Totally forgot to update the other instance of the Kafka Connector Cluster. Since it did not had the required replication.policy.class it did not work. After adding the jar to the other instance of the cluster it worked.

4

This topic was automatically closed 7 days after the last reply. New replies are no longer allowed.