Confluent JDBC Connector自定义Producer Interceptor未生效及日志问题排查

一、确保自定义拦截器生效的关键配置 #

1. 类路径与全限定类名 #

• 把你的CustomInterceptor打包成jar文件，放到Connect Worker的类路径下：
  • 如果你用的是Confluent Platform，通常放在share/java/kafka-connect-jdbc/目录（和JDBC Connector的jar放在一起），或者Connect的通用plugin目录share/java/plugins/。
  • 务必保证配置的producer.interceptor.classes是全限定类名，比如com.yourteam.interceptors.CustomInterceptor——只写CustomInterceptor的话，JVM找不到对应的类。

2. 正确配置拦截器属性 #

拦截器的配置分两种场景，按需选择：

• 全局生效（所有Connector共用）：在Connect Worker的核心配置文件（比如connect-distributed.properties或connect-standalone.properties）中添加：

  producer.interceptor.classes=com.yourteam.interceptors.CustomInterceptor
  

• 仅JDBC Connector生效：在提交JDBC Source Connector时的配置（比如JSON格式的配置文件）中添加带producer前缀的属性：

  {
    "name": "jdbc-source-connector",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
      "connection.url": "jdbc:mysql://localhost:3306/your_db",
      // 其他JDBC相关配置...
      "producer.interceptor.classes": "com.yourteam.interceptors.CustomInterceptor"
    }
  }
  

  注意：Connector级别的配置会覆盖Worker全局的producer配置。

3. 验证拦截器实现 #

确保你的CustomInterceptor正确实现了Kafka的ProducerInterceptor接口：

• 必须重写onSend(ProducerRecord)、onAcknowledgement(RecordMetadata, Exception)和close()三个核心方法。
• 检查有没有未捕获的初始化异常——如果拦截器启动失败，confluent log connect里会有ERROR级别的日志，先从这里排查问题。

二、让拦截器的INFO日志正常输出 #

默认情况下，Connect的日志配置不会打印自定义类的INFO级别日志，需要修改Log4j配置：

1. 找到Connect的Log4j配置文件：

   • Confluent Platform默认路径：standalone模式是etc/kafka/connect-log4j.properties，分布式模式（由Control Center管理）是etc/confluent-control-center/connect-log4j.properties。

2. 添加拦截器的日志规则：
   在文件末尾添加以下内容（替换为你的拦截器包路径）：

   # 把自定义拦截器的INFO日志输出到stdout
   log4j.logger.com.yourteam.interceptors=INFO, stdout
   # 避免日志重复输出到根日志
   log4j.additivity.com.yourteam.interceptors=false
   

   如果想把日志输出到单独文件，可以改成：

   log4j.logger.com.yourteam.interceptors=INFO, interceptorLog
   log4j.additivity.com.yourteam.interceptors=false
   log4j.appender.interceptorLog=org.apache.log4j.RollingFileAppender
   log4j.appender.interceptorLog.File=/var/log/confluent/kafka-connect/interceptor.log
   log4j.appender.interceptorLog.MaxFileSize=10MB
   log4j.appender.interceptorLog.MaxBackupIndex=5
   log4j.appender.interceptorLog.layout=org.apache.log4j.PatternLayout
   log4j.appender.interceptorLog.layout.ConversionPattern=%d{ISO8601} %p %c: %m%n
   

3. 重启Connect Worker：
   修改日志配置后必须重启Connect才能生效：

   confluent stop connect && confluent start connect
   

4. 查看日志：

   • 如果配置了输出到stdout，直接用confluent log connect就能看到拦截器的INFO日志。
   • 如果配置了单独文件，直接查看对应的日志文件即可。

内容的提问来源于stack exchange，提问作者SL101