Kafka Connect连接Cassandra时出现NoHostAvailableException异常求助

问题 #

尝试通过Kafka Connect将Cassandra表数据导入Kafka主题时，出现连接异常，堆栈信息如下：

org.apache.kafka.connect.errors.ConnectException: Couldn't connect to Cassandra.
    at io.lenses.streamreactor.connect.cassandra.source.CassandraSourceTask.start(CassandraSourceTask.scala:86)
    at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.initializeAndStart(AbstractWorkerSourceTask.java:280)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:259)
    at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:77)
    at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:236)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: com.datastax.driver.core.exceptions.NoHostAvailableException: All host(s) tried for query failed (tried: localhost/0:0:0:0:0:0:0:1:9042 (com.datastax.driver.core.exceptions.TransportException: [localhost/0:0:0:0:0:0:0:1:9042] Cannot connect), localhost/127.0.0.1:9042 (com.datastax.driver.core.exceptions.TransportException: [localhost/127.0.0.1:9042] Cannot connect))
    at com.datastax.driver.core.ControlConnection.reconnectInternal(ControlConnection.java:270)
    at com.datastax.driver.core.ControlConnection.connect(ControlConnection.java:109)
    at com.datastax.driver.core.Cluster$Manager.negotiateProtocolVersionAndConnect(Cluster.java:1813)
    at com.datastax.driver.core.Cluster$Manager.init(Cluster.java:1726)
    at com.datastax.driver.core.Cluster.init(Cluster.java:214)
    at com.datastax.driver.core.Cluster.connectAsync(Cluster.java:387)
    at com.datastax.driver.core.Cluster.connect(Cluster.java:338)
    at io.lenses.streamreactor.connect.cassandra.CassandraConnection$.getSession(CassandraConnection.scala:93)
    at io.lenses.streamreactor.connect.cassandra.CassandraConnection$.apply(CassandraConnection.scala:39)
    at io.lenses.streamreactor.connect.cassandra.source.CassandraSourceTask.$anonfun$start$2(CassandraSourceTask.scala:82)
    at scala.util.Try$.apply(Try.scala:210)
    at io.lenses.streamreactor.connect.cassandra.source.CassandraSourceTask.start(CassandraSourceTask.scala:82)
    ... 10 more

使用的Cassandra源连接器配置：

{
  "connector.class": "io.lenses.streamreactor.connect.cassandra.source.CassandraSourceConnector",
  "connect.cassandra.key.space": "my_keyspace",
  "connect.cassandra.contact.points": "localhost",
  "tasks.max": "1",
  "connect.cassandra.port": "9042",
  "connect.cassandra.kcql": "INSERT INTO cas_topic1 SELECT * FROM my_table",
  "connect.cassandra.password": "cassandra_password",
  "connect.cassandra.poll.interval.ms": "1000",
  "connect.cassandra.username": "cassandra_user"
}

排查与解决步骤 #

• 确认Cassandra服务状态
  先检查Cassandra是否正常运行：

  • 执行命令 nodetool status，若返回节点正常状态则服务运行；若命令报错，说明服务未启动，需启动Cassandra（如systemctl start cassandra或对应启动脚本）。
  • 用 telnet localhost 9042 或 nc -zv localhost 9042 测试端口连通性，连接失败则说明服务未监听该端口。

• 检查Cassandra监听配置
  打开Cassandra配置文件（通常为cassandra.yaml），确认以下配置：

  • listen_address：设置为Kafka Connect可访问的IP，同一机器可设为127.0.0.1或localhost，需与连接器配置的contact.points一致。
  • native_transport_port：确认值为9042，与连接器配置的端口匹配。
  • rpc_address：旧版驱动可能依赖该配置，建议设为0.0.0.0或具体可访问IP。
    修改配置后需重启Cassandra服务。

• 验证认证信息与权限

  • 确认connect.cassandra.username和connect.cassandra.password与Cassandra中创建的用户一致。
  • 检查该用户是否拥有my_keyspace和my_table的访问权限，执行CQL命令：

    LIST PERMISSIONS OF cassandra_user ON KEYSPACE my_keyspace;
    

    若权限不足，执行授权命令：

    GRANT SELECT ON KEYSPACE my_keyspace TO cassandra_user;
    

• 排查防火墙与网络策略

  • 若Kafka Connect与Cassandra不在同一机器，确认Cassandra所在机器的防火墙已开放9042端口。
  • 同一机器环境下，检查本地防火墙、SELinux是否阻止端口访问，可临时关闭测试。

• 确认驱动与Cassandra版本兼容性
  Lenses StreamReactor的Cassandra连接器使用的DataStax驱动需与Cassandra服务器版本匹配：

  • Cassandra 4.x对应DataStax Driver 4.x，Cassandra 3.x对应Driver 3.x。
    版本不匹配时，需更换对应版本的连接器插件。

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