kafka版本不一致导致的一个小问题（二）

绝地灬酷狼 2022-06-04 04:48 274阅读 0赞

背景介绍：

我们公司的实时流项目现在用的spark streaming比较多，这里在介绍下版本：

spark streaming2.1.0

kafka 0.9.0.0

spark streaming如果想要集成kafka使用时，必须得使用spark官网提供的依赖包，目前有两种大的kafka分支依赖集成包，他们的一些信息如下：

<table style="margin-top:0px; margin-bottom:16px; border-collapse:collapse; border-spacing:0px; display:block; width:622px; overflow:auto; word-break:keep-all; color:rgb(51,51,51); font-family:&quot;Helvetica Neue&quot;,Helvetica,&quot;Segoe UI&quot;,Arial,freesans,sans-serif,&quot;Apple Color Emoji&quot;,&quot;Segoe UI Emoji&quot;,&quot;Segoe UI Symbol&quot;; font-size:16px; line-height:25.6px; widows:1"> 
 <thead style=""> 
  <tr style="border-top:1px solid rgb(204,204,204)"> 
   <th style="padding:6px 13px; border:1px solid rgb(221,221,221)">描述信息</th> 
   <th style="padding:6px 13px; border:1px solid rgb(221,221,221)">spark-streaming-kafka-0-8</th> 
   <th style="padding:6px 13px; border:1px solid rgb(221,221,221)">spark-streaming-kafka-0-10</th> 
  </tr> 
 </thead> 
 <tbody style=""> 
  <tr style="border-top:1px solid rgb(204,204,204)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> kafka版本</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> 0.8.2.1 or higher</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> 0.10.0 or higher</td> 
  </tr> 
  <tr style="border-top-width:1px; border-top-style:solid; border-top-color:rgb(204,204,204); background-color:rgb(248,248,248)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> 稳定程度</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Stable</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Experimental</td> 
  </tr> 
  <tr style="border-top:1px solid rgb(204,204,204)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> 语言支持</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Scala, Java, Python</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Scala, Java</td> 
  </tr> 
  <tr style="border-top-width:1px; border-top-style:solid; border-top-color:rgb(204,204,204); background-color:rgb(248,248,248)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Receiver DStream</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Yes</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> No</td> 
  </tr> 
  <tr style="border-top:1px solid rgb(204,204,204)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Direct DStream</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Yes</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Yes</td> 
  </tr> 
  <tr style="border-top-width:1px; border-top-style:solid; border-top-color:rgb(204,204,204); background-color:rgb(248,248,248)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> SSL TLS Support</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> No</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Yes</td> 
  </tr> 
  <tr style="border-top:1px solid rgb(204,204,204)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Offset Commit Api</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> No</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Yes</td> 
  </tr> 
  <tr style="border-top-width:1px; border-top-style:solid; border-top-color:rgb(204,204,204); background-color:rgb(248,248,248)"> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Dynamic Topic Subscription</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> No</td> 
   <td style="padding:6px 13px; border:1px solid rgb(221,221,221); word-break:break-all"> Yes</td> 
  </tr> 
 </tbody> 
</table>

从上面的表格可以看出 spark-streaming-kafka-0-8目前是支持版本大于或等于0.8.2.1时需要用到的，因为我们生产环境的kafka的版本是0.9.0.0所以只能选择spark-streaming-kafka-0-8\_2.11这个依赖，然后spark streaming流程序跑起来，通过一定间隔不断从kafka消费数据，实时处理，整个流程是没有问题的，后来因为需要统一收集流程序的log中转到kafka中，最后通过logstash再发送到ElasticSearch中方便查看和检索日志，所以给项目集成了 kafka-log4j-appender-0.9.0.0，其功能是把log4j打印的日志给发送到kafka，配置完成之后再次启动项目，发现log也能收集到kafka中了，但通过后台发现时不时的会出现几条下面的log：

[2017-11-28 23:49:37,176] [WARN] [kafka-producer-network-thread | producer-2] Error in I/O with 192.168.10.160
    java.io.EOFException
    	at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:62)
    	at org.apache.kafka.common.network.Selector.poll(Selector.java:248)
    	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:192)
    	at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:191)
    	at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:122)
    	at java.lang.Thread.run(Thread.java:745)

但并不影响正常功能使用，从log里面能够看出来是生产者的问题，也就是说发送消息到kafka的server时出现连接中断了，导致抛出EOF异常。

那么为什么会中断连接呢？经查资料发现，这是由于kafka的版本不一致导致的，也就是说用0.8.2.1的kafka client向kafka0.9.0.0的server端发送数据，如果在经过了一定时间内，连接还没断开，那么服务端会主动断开这个连接，如果都是0.9.0.0的版本，服务端主动断开连接，客户端是不会抛出异常的，但由于版本不一样，在服务端主动中断的时候，就出现了上面的异常。

如何模拟重现？

（1）搭建一套0.9.0.0的kafka集群，为了方便重现，将server.properties里面的加上这个空闲连接关闭参数connections.max.idle.ms为30秒，默认不设置是10分钟，然后启动集群。

（2）在java项目里面使用0.8.2.1的client作为生产者，并使用生产者发送一条数据后，程序主动sleep40秒。

（3）然后观察等到30秒的时候就会抛出这个异常，但是主程序还是会等到40秒后结束，因为kafka发送消息是起的单独的线程所以抛出这个log时候主线程是不会受到影响的。

如何解决：

（1）最简单的办法就是升级client和server版本一致

（2）网上有朋友建议调大connections.max.idle.ms这个参数，减少抛出异常的次数，算是治标不治本吧，不建议这么干。

那么可能有朋友疑问，如果客户端一直不关闭空闲连接，必须得10分钟后由服务端强制关闭，那么会不会出现这个时间内kafka的连接资源被耗尽的情况呢？答案是几乎不可能，因为kafka允许每台主机默认的连接数为Int.MaxValue差不多21亿多吧。只要10分钟内每台主机的连接数达不到这个量级，程序就不会有问题。而实际情况生产者也不能出现这么多连接，所以我们的一些生产者程序一旦启动起来基本上不会调用close方法，除非在手动停止程序时，可以通过钩子函数来触发资源关闭，其他情况的空闲连接，可以由服务端进行管理通过超时关闭。注意如果是一直被占用的连接，服务端是不会主动关闭的，另外经过测试发现消费者就算版本不一致也不存在这个问题，目前来看只会版本不一致 而且是在生产者的程序中才会出现这个问题。