无法通过JDBC源连接器触发自定义生成器拦截器

Question

我已经创建了一个自定义的生产者拦截器（Audit病人拦截器），它接受一些自定义配置（application_id，类型等）。）。我已经从奥迪生产者拦截器项目生成了一个罐子，并将罐子放在Kafka连接中 /usr/share/java/monitoring-interceptors.当我尝试发布具有以下配置的JDBC-Source连接器时，我的审计拦截器没有被触发。

{
"name": "jdbc-source-xx-xxxx-xxx-xxx",
"config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:sqlserver://{{ip}}:1433;databaseName=XX;useNTLMv2=true",
    "connection.user": "SA",
    "connection.password": "Admin1234",
    "producer.interceptor.classes": "com.optum.payer.common.kafka.audit.interceptor.AuditProducerInterceptor",
    "topic.prefix": "MyTestTopic",
    "query": "SELECT ID, chart_id, request_id, UpdatedDate FROM xxx.xxx WITH (NOLOCK)",
    "mode": "timestamp",
    "timestamp.column.name": "UpdatedDate",
    "producer.audit.application.id": "HelloApplication",
    "producer.audit.type": "test type",
    "poll.interval.ms": "10",
    "tasks.max": "1",
    "batch.max.rows": "100",
    "validate.non.null": "false",
    "numeric.mapping":"best_fit",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://{{ip}}:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://{{ip}}:8081"
    
}}

正如您在配置中看到的，我在连接器配置中添加了以下道具来触发自定义拦截器。但我在Kafka Connect中没有看到任何与AuditProducerInterceptor相关的日志。

"producer.interceptor.classes": "com.optum.payer.common.kafka.audit.interceptor.AuditProducerInterceptor"
"producer.audit.application.id": "HelloApplication",
"producer.audit.type": "test type"

我尝试在 kafka-connect 配置中添加这三个配置，我能够触发拦截器。但是我想通过 JDBC 源连接器触发拦截器，以便我可以通过连接器传递自定义 props（application_id、类型等）。请帮我解决这个问题

Answer 1

如果在Connect worker中允许了客户端覆盖(默认情况下启用),您将需要使用< code>producer.override前缀

来自文档

从2.3.0开始，可以通过分别使用Kafka源或Kafka接收器的前缀producer.override.和consumer.override.为每个连接器单独配置客户端配置覆盖。