使用Kafka Connect InfluxDBSinkConnector时遭遇授权错误求助

问题 #

使用InfluxDB v1.7版本时，遇到Kafka Connect的InfluxDBSinkConnector连接AWS Timestream（兼容InfluxDB）的未授权错误。

提交连接器配置的curl命令：

curl -X POST -d @influxdb-sink-connector-dev.json http://localhost:8083/connectors -H "Content-Type: application/json" | jq .

连接器配置内容：

{
  "name": "InfluxDBSinkDemoConnector",
  "config": {
    "connector.class": "io.confluent.influxdb.InfluxDBSinkConnector",
    "tasks.max": "1",
    "topics": "tenant-health-metrics",
    "influxdb.url": "https://<aws-timestream-url>.timestream-influxdb.ap-south-1.on.aws:8086",
    "influxdb.db": "influx-dev",
    "influxdb.username": "<username>",
    "influxdb.password": "<password>",
    "measurement.name.format": "${topic}",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://localhost:8081",
    "name": "InfluxDBSinkDemoConnector"
  },
  "tasks": [],
  "type": "sink"
}

注：原配置中influxdb.url和value.converter.schema.registry.url未加引号，属于JSON语法错误，已修正。

启动连接器时出现的未授权错误日志：

[2025-09-16 20:57:32,343] ERROR [InfluxDBSinkDemoConnector|worker] WorkerConnector{id=InfluxDBSinkDemoConnector} Error while starting connector (org.apache.kafka.connect.runtime.WorkerConnector:227)
org.influxdb.InfluxDBException: {"code":"unauthorized","message":"Unauthorized"}
                at org.influxdb.InfluxDBException.buildExceptionForErrorState(InfluxDBException.java:175) ~[influxdb-java-2.24.jar:2.24]
                at org.influxdb.impl.InfluxDBImpl.execute(InfluxDBImpl.java:846) ~[influxdb-java-2.24.jar:2.24]
                at org.influxdb.impl.InfluxDBImpl.executeQuery(InfluxDBImpl.java:833) ~[influxdb-java-2.24.jar:2.24]
                at org.influxdb.impl.InfluxDBImpl.describeDatabases(InfluxDBImpl.java:759) ~[influxdb-java-2.24.jar:2.24]
                at io.confluent.influxdb.InfluxDBSinkConnector.start(InfluxDBSinkConnector.java:52) ~[kafka-connect-influxdb-1.2.11.jar:?]
                at org.apache.kafka.connect.runtime.WorkerConnector.doStart(WorkerConnector.java:219) ~[connect-runtime-8.0.0-ce.jar:?]
                at org.apache.kafka.connect.runtime.WorkerConnector.start(WorkerConnector.java:248) ~[connect-runtime-8.0.0-ce.jar:?]
                at org.apache.kafka.connect.runtime.WorkerConnector.doTransitionTo(WorkerConnector.java:408) ~[connect-runtime-8.0.0-ce.jar:?]
                at org.apache.kafka.connect.runtime.WorkerConnector.doTransitionTo(WorkerConnector.java:389) ~[connect-runtime-8.0.0-ce.jar:?]
                at org.apache.kafka.connect.runtime.WorkerConnector.doRun(WorkerConnector.java:168) ~[connect-runtime-8.0.0-ce.jar:?]
                at org.apache.kafka.connect.runtime.WorkerConnector.run(WorkerConnector.java:128) ~[connect-runtime-8.0.0-ce.jar:?]
                at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$7(Plugins.java:356) ~[connect-runtime-8.0.0-ce.jar:?]
                at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572) ~[?:?]
                at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317) ~[?:?]
                at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) ~[?:?]
                at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) ~[?:?]
                at java.base/java.lang.Thread.run(Thread.java:1583) [?:?]

已知信息：可正常通过浏览器访问AWS Timestream URL并使用用户名密码登录。

可能的原因及解决方法 #

1. JSON配置语法错误 #

原配置中influxdb.url和value.converter.schema.registry.url的值未用双引号包裹，属于无效JSON语法。提交时Kafka Connect可能解析失败，导致配置未正确加载，进而使用默认值或空值认证，引发未授权错误。

• 解决方法：确保所有字符串类型的配置值都用双引号包裹，参考修正后的配置格式。

2. 连接器与Timestream兼容性问题 #

使用的kafka-connect-influxdb-1.2.11.jar和influxdb-java-2.24.jar可能未适配AWS Timestream的InfluxDB兼容接口认证逻辑。Timestream的兼容模式可能要求特定的认证头或参数格式，旧版本客户端未支持。

• 解决方法：升级InfluxDB Java客户端到最新兼容版本，或检查Confluent连接器是否有针对AWS Timestream的更新版本；若使用IAM认证，需替换为AWS官方适配的连接方式。

3. 认证方式不匹配 #

AWS Timestream的InfluxDB兼容模式支持多种认证方式（用户名密码、AWS IAM凭证），但连接器默认的认证流程可能与Timestream配置不符。浏览器登录使用的会话认证，与Java客户端的Basic Auth流程存在差异，可能导致认证失败。

• 解决方法：
  • 确认Timestream是否开启Basic Auth，若使用IAM认证，需在连接器配置中添加AWS IAM参数（如influxdb.aws.region、aws.access.key.id、aws.secret.access.key）。
  • 用curl模拟连接器的认证请求，测试权限：

    curl -u "<username>:<password>" https://<aws-timestream-url>.timestream-influxdb.ap-south-1.on.aws:8086/query?q=SHOW+DATABASES
    

    若请求返回未授权，说明用户名密码的权限或格式存在问题；若成功，说明连接器客户端实现存在差异。

4. 数据库权限不足 #

虽然能登录Timestream，但当前用户可能没有influx-dev数据库的访问权限（如READ或列出数据库的权限），连接器启动时调用describeDatabases接口会触发未授权错误。

• 解决方法：检查Timestream中该用户的权限配置，确保其拥有目标数据库的访问权限，包括列出数据库的权限。

5. 网络或代理问题 #

Kafka Connect所在环境可能无法正常访问Timestream URL，或代理修改了请求头导致认证信息未正确传递。

• 解决方法：在Kafka Connect服务器上用telnet或curl测试网络连通性，确认能访问Timestream的8086端口；若使用代理，需在连接器配置中添加代理参数，确保认证信息正确传递。

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