Reputation: 537
I'm new to Kafka and I can't figure out how to use "transforms.router.type" to make it work with my Debezium setup. So I made special event transformation java class and prepeared configuration for it to deploy to container which looks like this:
curl -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d
{
"name": "task-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"tasks.max": "1",
"slot.name" : "task_engine_saga",
"database.hostname": "postgres",
"database.port": "5432",
"database.user": "postgres",
"database.password": "postgres",
"database.dbname" : "tasks",
"schema.whitelist": "public",
"table.whitelist" : "public.task",
"tombstones.on.delete" : "false",
"transforms" : "router",
"transforms.router.type" : "com.task.connect.TaskEventRouter"
}
}
And the response says that this config can't be found.
CREATE kafka connector task-connector....
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
100 1091 100 516 100 575 8322 9274 --:--:-- --:--:-- --:--:-- 17596{"error_code":400,"message":"Connector configuration is invalid and contains the following 3 error(s):\nInvalid value com.task.connect.TaskEventRouter for configuration transforms.router.type: Class com.task.connect.TaskEventRouter could not be found.\nInvalid value null for configuration transforms.router.type: Not a Transformation\nA value is required\nYou can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`"}
Then I copied to container /connect directory my host folder with jar file in which I have my java class with event transformation logic, but it didn't help too. Could someone please help me and tell what should I do to make this custom transforms.router.type work with my Debezium set up?
My containers docker-compose set up:
version: '3'
services:
pgadmin:
container_name: pgadmin_container
image: dpage/pgadmin4
environment:
PGADMIN_DEFAULT_EMAIL: ${PGADMIN_DEFAULT_EMAIL:[email protected]}
PGADMIN_DEFAULT_PASSWORD: ${PGADMIN_DEFAULT_PASSWORD:-admin}
volumes:
- pgadmin:/root/.pgadmin
ports:
- "${PGADMIN_PORT:-5050}:80"
restart: unless-stopped
zookeeper:
image: debezium/zookeeper:1.3
ports:
- 2181:2181
- 2888:2888
- 3888:3888
kafka:
image: debezium/kafka:1.3
ports:
- 9092:9092
links:
- zookeeper
environment:
- ZOOKEEPER_CONNECT=zookeeper:2181
postgres:
image: debezium/example-postgres:1.3
ports:
- 5432:5432
environment:
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=postgres
- PGDATA=/data/postgres
- POSTGRES_DB=${POSTGRES_DB:-task_engine}
connect:
image: debezium/connect:1.3
ports:
- 8083:8083
links:
- kafka
- postgres
environment:
- BOOTSTRAP_SERVERS=kafka:9092
- GROUP_ID=1
- CONFIG_STORAGE_TOPIC=my_connect_configs
- OFFSET_STORAGE_TOPIC=my_connect_offsets
- STATUS_STORAGE_TOPIC=my_connect_statuses
volumes:
postgres:
pgadmin:
Upvotes: 1
Views: 2112
Reputation: 191874
I made special event transformation java class
From what I can tell, you have not mounted this compiled class into your debezium image or edited the plugin.path
file to include that JAR
If you have created your own Docker image with that data, then you should change image: debezium/connect:1.3
This would explain why the transform is not available
Upvotes: 0