Kafka配置内外网(NAT)访问小记

Kafka集群的内外网架构图如下：

背景：

    刚开始kafka集群只提供内网访问，后面因为业务需求，需要提供kafka集群外网访问。但是由于内网机器没有公网IP，于是通过NAT网关映射端口的方式提供公网IP访问。这里之所以中间有一台Nginx代理是因为公司只给这个网段区域的服务器授权可以做NAT的映射，然后通过其代理到内网。这样既保证了安全性，也提高了可配置性。Nginx配置如下：

    upstream kafka_01{        server 10.192.39.33:9092;    }    upstream kafka_02{        server 10.192.39.35:9092;    }    upstream kafka_03{        server 10.192.39.37:9092;    }
    server {        listen 9091;        access_log  /var/log/nginx/xpukafka.log proxy;        error_log  /var/log/nginx/xpukafka-error.log;        proxy_pass kafka_01;        proxy_connect_timeout 8s;        proxy_timeout 24h;    }    server {        listen 9092;        access_log  /var/log/nginx/xpukafka.log proxy;        error_log  /var/log/nginx/xpukafka-error.log;        proxy_pass kafka_02;        proxy_connect_timeout 8s;        proxy_timeout 24h;    }    server {        listen 9093;        access_log  /var/log/nginx/xpukafka.log proxy;        error_log  /var/log/nginx/xpukafka-error.log;        proxy_pass kafka_03;        proxy_connect_timeout 8s;        proxy_timeout 24h;    }

问题：

    Kafka集群只监听了9092端口，没有开启9094端口的监听。配置完nginx代理（反向代理到9092端口）和NAT映射后在阿里云测试端口是通的，并且kafka的外网broker地址也是可以连接成功的。但是无法发送消息，发送消息时报错如下：

分析：

    通过在阿里云ECS上使用python测试连接kafka公网地址并发送消息

    发现确实是能连接上，但是发送不了消息。最终会timeout。此时的kafka配置如下：

broker.id=2listeners=PLAINTEXT://:9092advertised.listeners=PLAINTEXT://10.192.39.33:9092num.network.threads=9num.io.threads=16socket.send.buffer.bytes=1048576socket.receive.buffer.bytes=1048576replica.socket.receive.buffer.bytes=1048576socket.request.max.bytes=524288000log.dirs=/xpu-k8s-data/data/kafka/kafka9094/datanum.partitions=5num.recovery.threads.per.data.dir=8offsets.topic.replication.factor=3transaction.state.log.replication.factor=3transaction.state.log.min.isr=2log.retention.hours=72log.segment.bytes=1073741824log.retention.check.interval.ms=300000zookeeper.connect=10.192.39.33:2184,10.192.39.35:2184,10.192.39.37:2184zookeeper.connection.timeout.ms=10000zookeeper.session.timeout.ms=10000group.initial.rebalance.delay.ms=3000default.replication.factor=2message.max.bytes=10485760replica.fetch.max.bytes=10485760unclean.leader.election.enable=falsenum.replica.fetchers=4

    注意关键配置参数listeners和advertised.listeners。这里只配置了监听9092端口。

    确认Kafka，Nginx和NAT配置无误后，修改了代码的超时时间验证还是无法发送消息。于是进行抓包分析（抓包文件已经删除，无法截图看到）发现：在阿里云访问kafka公网地址时，连接成功，但是在发送消息时，程序又用kafka的内网地址10.192.39.x和9092端口去连接，导致连接超时无法发送消息。查阅资料得知，在往kafka发送消息时，kafka会从zk查询advertised.listeners配置的地址返回给生产者，告知其用这个地址发送消息。所以要解决这个问题就需要把advertised.listeners地址配置为公网地址。但是这样又会带来一个问题，这样所有内网访问都要走公网绕出去再回来，不科学！

解决：

    查看kafka的配置得知，kafka是支持配置内外网地址的。配置如下：

broker.id=2listeners=INSIDE://:9092,OUTSIDE://:9094inter.broker.listener.name=INSIDEadvertised.listeners=INSIDE://10.192.39.33:9092,OUTSIDE://120.xxx.xxx.217:19091listener.security.protocol.map=INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXTnum.network.threads=9num.io.threads=16socket.send.buffer.bytes=1048576socket.receive.buffer.bytes=1048576replica.socket.receive.buffer.bytes=1048576socket.request.max.bytes=524288000log.dirs=/xpu-k8s-data/data/kafka/kafka9094/datanum.partitions=5num.recovery.threads.per.data.dir=8offsets.topic.replication.factor=3transaction.state.log.replication.factor=3transaction.state.log.min.isr=2log.retention.hours=72log.segment.bytes=1073741824log.retention.check.interval.ms=300000zookeeper.connect=10.192.39.33:2184,10.192.39.35:2184,10.192.39.37:2184zookeeper.connection.timeout.ms=10000zookeeper.session.timeout.ms=10000group.initial.rebalance.delay.ms=3000default.replication.factor=2message.max.bytes=10485760replica.fetch.max.bytes=10485760unclean.leader.election.enable=falsenum.replica.fetchers=4

    就是修改listeners和advertised.listeners参数，区分INSIDE和OUTSIDE，然后添加inter.broker.listener.name和listener.security.protocol.map参数。修改后重启kafka集群，kafka会同时监听9092和9094两个端口。

    修改nginx反向代理端口为9094。内网连接的地址不变，公网连接的地址也不变。测试内外网发送消息均正常。