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