How to configure 2 kafka cluster bootstrap server for same kafka topic and table

[shubhamshinde7]

I have 2 kafka clusters with same topic name as table name. I created below config using API. Now I am getting │ valk-kafka-connect [2024-09-25 19:19:26,281] WARN Sending [20] records to DLQ for exception: Topic: [sna.valk.sn_valk.user_ │ │ data], Partition: [0], MinOffset: [125], MaxOffset: [144], (QueryId: [994c936b-19eb-4d25-9964-8b48ae28a2f3]) (com.clickhous │ │ e.kafka.connect.sink.ClickHouseSinkTask) error every time

Below is kafka connect config
`
curl -k -X PUT -H "Content-Type: application/json"
--data '{
"consumer.override.bootstrap.servers": "kafka1:3000",
"consumer.override.group.id": "sna.valk.sn_valk.user_data_shubham-group",
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
"consumer.override.max.poll.records": 100000,
"consumer.override.fetch.max.wait.ms": 60000,
"consumer.override.fetch.min.bytes": 200615200,
"consumer.override.request.timeout.ms": 75000,
"consumer.override.max.partition.fetch.bytes": 200615200,
"tasks.max": "4",
"topics": "sna.valk.sn_valk.user_data",
"clickhouse.table": "sna.valk.sn_valk.user_data",
"ssl": "true",
"security.protocol": "SSL",
"hostname": "valk-clickhouse-svc",
"port": "8443",
"database": "valk",
"errors.retry.timeout": "60",
"exactlyOnce": "true",
"value.converter.schemas.enable": "false",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"username": "default",
"schemas.enable": "false",
"errors.deadletterqueue.context.headers.enable": "true",
"errors.deadletterqueue.topic.name": "sna.valk.sn_valk.user_data_dlq",
"errors.tolerance": "all",
"clickhouseSettings": "date_time_input_format=best_effort"
}'
https://valk-kafka-connect-svc:8443/connectors/valk-shubham/config

curl -k -X PUT -H "Content-Type: application/json"
--data '{
"consumer.override.bootstrap.servers": "kafka2:3000",
"consumer.override.group.id": "sna.valk.sn_valk.user_data_release-group",
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
"consumer.override.max.poll.records": 100000,
"consumer.override.fetch.max.wait.ms": 60000,
"consumer.override.fetch.min.bytes": 200615200,
"consumer.override.request.timeout.ms": 75000,
"consumer.override.max.partition.fetch.bytes": 200615200,
"tasks.max": "4",
"topics": "sna.valk.sn_valk.user_data",
"clickhouse.table": "sna.valk.sn_valk.user_data",
"ssl": "true",
"security.protocol": "SSL",
"hostname": "valk-clickhouse-svc",
"port": "8443",
"database": "valk",
"keeperOnCluster": "ON CLUSTER valk_global",
"errors.retry.timeout": "60",
"exactlyOnce": "true",
"value.converter.schemas.enable": "false",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"username": "default",
"schemas.enable": "false",
"errors.deadletterqueue.context.headers.enable": "true",
"errors.deadletterqueue.topic.name": "sna.valk.sn_valk.user_data_dlq",
"errors.tolerance": "all",
"clickhouseSettings": "date_time_input_format=best_effort"
}'
https://valk-kafka-connect-svc:8443/connectors/valk-release/config

`

[Paultagoras]

Could you paste more of the logs?

[shubhamshinde7]

i truncated connect_state table that didn't help

[Paultagoras]

@shubhamshinde7 Please try using the latest version of the connector (as of now, 1.2.3) and see if that resolves the issue. That said, if you don't need exactly-once I would recommend setting "exactlyOnce": "false"

[shubhamshinde7]

i updated connector to latest version and created two separate connector with different group id and offset and status topics it worked for me

[shubhamshinde7]

Now i am facing issue with inserting array of json in table. In logs I am not seeing any errors. Below is table and json structure

CREATE TABLE IF NOT EXISTS valk.user_properties_local ON CLUSTER '{global_cluster}' (
InstanceId LowCardinality(String),
InstanceUserId FixedString(64),
LastModified SimpleAggregateFunction(max, DateTime('UTC')),
LatestUserGroups AggregateFunction(argMax, Array(LowCardinality(String)), DateTime('UTC')),
LatestUserRoles AggregateFunction(argMax, Array(LowCardinality(String)), DateTime('UTC')),
LatestSysUser AggregateFunction(argMax, String, DateTime('UTC'))
)
ENGINE = AggregatingMergeTree()
ORDER BY (InstanceId, InstanceUserId);

{
"result":[
{
"InstanceId":"abc1",
"InstanceUserId":"shubham.shinde",
"UpdateTime":"1727678916",
"UserGroups":["group1","group2"],
"UserRoles":["role1","role2"],
"SysUser":{
"dept":"eng",
"title":"sde"
}
},
{
"InstanceId":"abc2",
"InstanceUserId":"shubham.shinde",
"UpdateTime":"1727678916",
"UserGroups":["group1","group2"],
"UserRoles":["role1","role2"],
"SysUser":{
"dept":"eng",
"title":"sde"
}
}
]
}

[Paultagoras]

What behavior are you seeings with that data?