草庐IT

【kafka】Connection to node -1) terminated during authentication. This may happen due to any of

kiraraLou 2023-04-21 原文

前言

最近在使用 java 消费 kafka 服务时，遇到报错，报错内容如下：

环境信息

  • kafka 2.5.0
  • kerberos

报错内容

Connection to node -1) terminated during authentication. This may happen due to any of the following reasons: (1) Authentication failed due to invalid credentials with brokers older than 1.0.0, (2) Firewall blocking Kafka TLS traffic (eg it may only allow HTTPS traffic), (3) Transient network issue.

[2020-05-18 18:14:38,615] INFO [Group Metadata Manager on Broker 193]: Removed 0 expired offsets in 0 milliseconds. (kafka.coordinator.GroupMetadataManager)
[2020-05-18 18:24:38,615] INFO [Group Metadata Manager on Broker 193]: Removed 0 expired offsets in 0 milliseconds. (kafka.coordinator.GroupMetadataManager)
[2020-05-18 18:34:38,616] INFO [Group Metadata Manager on Broker 193]: Removed 0 expired offsets in 1 milliseconds. (kafka.coordinator.GroupMetadataManager)
[2020-05-18 18:44:02,879] WARN Unexpected error from /134.84.53.60; closing connection (org.apache.kafka.common.network.Selector)
org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = -720899)
at org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:89)
at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:160)
at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:141)
at org.apache.kafka.common.network.Selector.poll(Selector.java:286)
at kafka.network.Processor.run(SocketServer.scala:413)
at java.lang.Thread.run(Thread.java:748)
[2020-05-18 18:44:38,615] INFO [Group Metadata Manager on Broker 193]: Removed 0 expired offsets in 0 milliseconds. (kafka.coordinator.GroupMetadataManager)
[2020-05-18 18:54:38,615] INFO [Group Metadata Manager on Broker 193]: Removed 0 expired offsets in 0 milliseconds. (kafka.coordinator.GroupMetadataManager)

排查方向

  1. 时钟同步（应用与kafka集群之间、kafka集群内部之间）重要
  2. 使用主机名是否与KDC、KAFKA服务器网络/端口正确通信。
  3. 防火墙是否关闭，检查selinux。
  4. krb5.conf kafka_client_jaas.conf xxx.keytab 三个配置文件内容是否正确，文件所属用户，权限是否正确。
  5. 是否指定 JVM 参数，或者在代码里面是否指定环境变量。
  6. 如下配置，是否正确配置
    properties.setProperty("sasl.kerberos.service.name", "kafka");
    properties.setProperty("sasl.mechanism", "GSSAPI");
    properties.setProperty("security.protocol", "SASL_PLAINTEXT");
  7. Kafka 相关jar包版本是否一致，是否有依赖冲突。
  8. 与 KDC 服务器网络是否通信良好（有无丢包），因为kerberos 默认是udp通信，建议换成tcp。

对于使用spark-kafka 程序消费 kafka 的，需要注意 "sasl.kerberos.service.name”，“sasl.mechanism”， “security.protocol” 三项配置需要增加前缀 “kafka.”

补充

另外查到网上有别的解决方式：

  1. 调整 socket.request.max.bytes 参数
  2. 缺少jackjson 的 pom 依赖

目前个人感觉，上面两个应该不是正确的解决方向，这里也顺便贴出来，供大家参考。

authenticationConnectionspanclasstokenkafkajava分布式

有关【kafka】Connection to node -1) terminated during authentication. This may happen due to any of的更多相关文章

  1. kafka如何动态消费新增topic主题 - 2

    一、解决痛点使用spring-kafka客户端，每次新增topic主题，都需要硬编码客户端并重新发布服务，操作麻烦耗时长。kafkaListener虽可以支持通配符消费topic，缺点是并发数需要手动改并且重启服务。对于业务逻辑相似场景，创建新主题动态监听可以用kafka-batch-starter组件二、组件能力1、新增topic名称为：auto.topic1（由于配置spring.kafka.consumer.prefix为auto，因此只有auto前缀的topic，才会被组件动态监听。）2、应用输出日志，监听到新增auto.topic1，并初始化客户端(主题刷新间隔为10s)3、发新的消

  2. Zookeeper、Nacos、Dubbo、Kafka之间的关系 - 2

    1.Zookeeper　　Zookeeper是 ApacheHadoop 的子项目，是一个树型的目录服务，支持变更推送，适合作为Dubbo服务的注册中心，工业强度较高。　　Zookeeper的功能主要是它的树形节点来实现的。当有数据变化的时候或者节点过期的时候，会通过事件触发通知对应的客户端数据变化了，然后客户端再请求zookeeper获取最新数据，采用push-pull来做数据更新。服务注册和消费信息直接存储在zk树形节点上，集群下采用过半机制保证服务节点间一致性。　2.Nacos　　Nacos是 Alibaba 公司推出的开源工具，用于实现分布式系统的服务发现与配置管理。Nacos是Dub

  3. Spark Kafka流媒体 - 如何确定批次的末端 - 2

    我使用Kafka流媒体从KAFKA主题中消费。（KafkaDirect流）此主题中的数据每5分钟从另一个来源到达。现在，我需要处理每5分钟后到达的数据，并将其转换为SparkDataFrame。现在，流是数据的连续流。我的问题是，如何确定我已经完成了在Kafka主题中加载的第一组数据的阅读？（以便我可以将其转换为数据框架并开始我的工作）我知道我可以提及某个数字的批处理间隔（在JavastreamingContext中），但是即使那样，我也永远无法确定源将数据将数据推到主题的时间。欢迎任何建议。看答案如果我正确理解您的问题，您希望不创建批处理，直到阅读5分钟的所有数据。开箱即用的Spark不会提

  4. 流批一体计算引擎-4-[Flink]消费kafka实时数据 - 2

    Python3.6.9Flink1.15.2消费KafakaTopicPyFlink基础应用之kafka通过PyFlink作业处理Kafka数据1环境准备1.1启动kafka(1)启动zookeeperzkServer.shstart(2)启动kafkacd/usr/local/kafka/nohup./bin/kafka-server-start.sh./config/server.properties>>/tmp/kafkaoutput.log2>&1&或者./bin/kafka-server-start.sh-daemon./config/server0.properties(3)查看进

  5. javascript - 在 Web 浏览器上实时接收 Kafka 事件 - 2

    我们使用Kafka集群实时发送/接收消息。我们能够成功地向Kafka主题发布消息。现在我们希望在使用JavaScript的Web浏览器上运行的单页应用程序(SPA)上实时接收这些消息。是否可以直接从Kafka将消息推送到在任何著名浏览器上运行的最新版本的SPA？我找到了使用NodeJS实时接收消息的示例，但没有找到在Web浏览器上运行JavaScript的示例。 最佳答案 Kafka有Javascript客户端，但鉴于您的用例描述，我建议您在浏览器端javascript和Kafka之间使用REST或WebSocket代理。这将确保消

  6. go - 转义 json 对象以创建 kafka 轴突有效载荷 - 2

    我一直在尝试创建需要转义格式的json数据的负载。我可以序列化该对象，但不确定如何对同一对象进行双重转义？我是否应该双重编码(marshal)我的对象以便它逃脱它？Input:{"new":{"Id":"1","Class":"23"}}Expected:{\"new\":{\"Id\":\"1\",\"Class\":\"23\"}} 最佳答案 将最后一行更改为fmt.Printf("%q",string(b))-这会导致格式为“转义字符串”。(或者如果你想存储转义字符串，```fmt.Sprintf``)https://play

  7. json - Kafka - Json(最佳实践) - 2

    我需要将RESTAPI调用的输出推送到KAFKA。Restapi返回json输出，其中包含支持信息以及数据输出到json.RawMessagetypeResponsestruct{RequestIDstring`json:"requestId"`Successbool`json:"success"`NextPageTokenstring`json:"nextPageToken,omitempty"`MoreResultbool`json:"moreResult,omitempty"`Errors[]struct{Codestring`json:"code"`Messagestring`

  8. asynchronous - 具有异步 goroutines 的 Kafka 消费者 - 2

    我正在为我的消费者使用sarama(https://github.com/Shopify/sarama/)和Kafka0.8.0。这是我的代码的样子:consumerLoop:for{select{caseevent:=我正在使用缓冲channel(c.sem)来控制一次可以运行多少个processJobgoroutine。这就是我控制消费者的并发/速度的方式。我在使用这种方法时遇到的问题是，如果我需要更改并发性，我必须关闭使用者并重新启动它(channel缓冲区大小是一个命令行标志)。我记录了已处理的偏移量，我必须查看我的日志以确定处理了哪些偏移量以及我希望消费者从哪里恢复。我想要一

  9. go - 只有一个消费者获得 5 个分区的 kafka 消息 - 2

    我们使用Go的confluentkafka包测试了具有2和3个消费者的消费者组(知道我们将来可能会有更多消费者)。每个主题有10个分区，消息在所有消费者之间拆分。每个主题有5个分区，但不知何故只有一个消费者获取消息。知道为什么会出现这种行为吗？ 最佳答案 您可以尝试使用此命令查看分区分配情况，并将结果添加到此处吗？bin/kafka-consumer-groups.sh--bootstrap-server:9092--describe--group--members--verbose默认情况下，Kafka使用范围分区方案，因此有时会

  10. go - Kafka Golang 客户端无法连接到 Message Hub 服务 - 2

    这是我正在运行的代码片段:err:=godotenv.Load()iferr!=nil{log.Fatal("Errorloading.envfile")}broker:=os.Getenv("BROKER")topic:=os.Getenv("TOPIC")username:=os.Getenv("USERNAME")password:=os.Getenv("PASSWORD")calocation:=os.Getenv("CALOCATION")p,err:=kafka.NewProducer(&kafka.ConfigMap{"metadata.broker.list":brok

随机推荐