解决阿里云Flink连接Kafka报UnknownHostException的问题
问题描述
最近遇到一个较为麻烦的问题,写篇文章记录下解决问题的思路。
在使用阿里云Flink时,想连接公司之前自建的Kafka服务,但报UnknownHostException
。这是由于Kafka通过advertised.listeners
配置了域名,一般的解决办法是修改本机的/etc/hosts
文件,添加域名和对应的IP,这样就能解析成功。但棘手的地方在于阿里云的Flink是云端服务,无法修改/etc/hosts
,所以导致该Flink无法正常访问Kafka。后面我们也与公司运维及阿里技术支持进行了交流,阿里给出的解决方案是使用PrivateZone服务,提供一个私有DNS解析服务并绑定对应的VPC。然而阿里的PrivateZone服务也存在一个问题,它的域名只支持FQDN,也就是全路径的域名格式,但我们自建的Kakfa使用了简单Hostname形式,如:cx-kafka1
,cx-kafka2
等,这样导致阿里的方案也被pass了。该集群由于有不少地方在使用,因此配置不能修改,这个问题也被搁置了。最近花了些时间,顺着之前的思路解决了该问题。
原理及解决思路
- 我们知道
bootstrap.servers
配置的地址信息,只是Kafka用来获取连接引导信息的地址,是用于发现Kafka完整集群信息的,而连接真正Kafka服务的地址是advertised.listeners
广播到Zookeeper的。所以为了正常连接Kafka,客户端必须要能解析advertised.listeners
的地址。既然阿里的PrivateZone无法配置我们这样的域名格式,我们能否搭建自己的DNS Server解析该地址,并让阿里的Flink服务使用我们的DNS Server呢?
DNS Server很好办,由于没有大量的解析需求,我采用了短小精悍的dnsmasq,配置起来也很简单。
#dnsmasq config, for a complete example, see:
# http://oss.segetech.com/intra/srv/dnsmasq.conf
log-queries
address=/cx-kafka1/xxx.xxx.xxx.xxx
address=/cx-kafka2/xxx.xxx.xxx.xxx
.........
.........
测试一下:
然后,向Flink集群开放UDP 53端口,第一步就算OK了。
- 接下来,就是Flink服务如何去使用该DNS Server,通过调研我们了解到,JVM提供了两个参数
sun.net.spi.nameservice.provider.<n>
和sun.net.spi.nameservice.nameservers
,可以指定JVM解析域名的方式。通过在系统管理 > 作业模板 > Flink配置中增加下面的配置,将上面自建的DNS Server地址指定给JVM。
env.java.opts: >-
-Dsun.net.spi.nameservice.provider.1=default
-Dsun.net.spi.nameservice.provider.2=dns,sun
-Dsun.net.spi.nameservice.nameservers=xxx.xxx.xxx.xxx
由于JDK7之后是链式解析,因此默认的解析方式放在前面,如果默认的DNS解析失败就会使用后面我们自定义的DNS Server。经过测试,阿里的Flink正常连接上了我们自己Kafka集群。这里还有一点需要注意,如果上述方法没有生效,Per-Job集群需要新建作业,而Session集群也需要按上面描述重新配置下,这样新的JVM配置才会生效。
上述修改JVM自定义DNS解析的方式,其实也可以延伸到其他不方便配置hosts的环境,如Docker容器中。