kafka多生产者的情况下，如何确认是哪个生产者导致的写入不均衡？

我使用的是flink，生产者没有设置key，轮询发送的，使用了SSL。
经过测试，发现主要是Controller这一台的写入比较多，在使用无认证或sasl的时候没有这样的情况。
kafka端有这样的报错：

WARN Failed to send SSL Close message  (org.apache.kafka.common.network.SslTransportLayer)
java.io.IOException: Connection reset by peer

flink端有这样的报错

org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer [] - Error while sending record to Kafka: Expiring 104 record(s) for test-2:120000 ms has passed since batch creation

kafka的配置

listeners=SASL_PLAINTEXT://192.168.1.12:9097,PLAINTEXT://192.168.1.12:9092,SSL://192.168.1.12:9096
advertised.listeners=SASL_PLAINTEXT://192.168.1.12:9097,PLAINTEXT://192.168.1.12:9092,SSL://192.168.1.12:9096
ssl.keystore.location=/home/keystore.jks
ssl.keystore.password=***
ssl.key.password=***
ssl.truststore.location=/home/truststore.jks
ssl.truststore.password=***
#ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type=JKS 
ssl.truststore.type=JKS 

ssl.endpoint.identification.algorithm=

security.inter.broker.protocol=SASL_PLAINTEXT

sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

大神有什么思路吗？是kafka SSL的问题还是flink的问题呢？

SSL数据传输加密之后，kafka吞吐也随之变慢了，产生了一些消息等待发送超过了120000ms的错误。
直白点，就是kakfa集群达到了上限，可以增加kafka节点或者扩大kafka配置解决你的问题：
参考：https://www.orchome.com/9783

嗯嗯，这个我明白，我现在不理解的是，为啥用SSL的时候，就会对某几台有很大的压力；比如我图里有两台写入量是其他节点的两倍了。

用sasl的时候就没有这样的情况。

是不是其他的节点，ssl没生效。

都是生效的，我手动切换controller之后，压力就会转到新的controller上面去。

如果你能只放行部分生产者，那就通过查询与kafka节点的长连接，来确认生产者连的数量。

比如：

# 获取建立的长连接
netstat -unltpa|grep 9092

每个broker上面的连接数量我都确认了，数量上都是基本相同的，相差在1-2个。
整体资源使用也没有很高，40%左右(内存、IO、CPU)。