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

Posted 2021-05-02 我是攻城师

背景介绍：

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

spark streaming2.1.0

kafka 0.9.0.0

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

从上面的表格可以看出 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：

但并不影响正常功能使用，从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方法，除非在手动停止程序时，可以通过钩子函数来触发资源关闭，其他情况的空闲连接，可以由服务端进行管理通过超时关闭。注意如果是一直被占用的连接，服务端是不会主动关闭的，另外经过测试发现消费者就算版本不一致也不存在这个问题，目前来看只会版本不一致 而且是在生产者的程序中才会出现这个问题。