我正在使用KafkaPublish处理器将消息发布到使用SASL_SSL安全协议保护的kafka主题(我们不使用Kerberos)。
我得到以下错误:
WARN Failed to send SSL Close message
(org.apache.kafka.common.network.SslTransportLayer)
java.io.IOException: An existing connection was forcibly closed by the
remote host
at sun.nio.ch.SocketDispatcher.write0(Native Method)
at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:51)
at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93)
at sun.nio.ch.IOUtil.write(IOUtil.java:65)
at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:471)
at
org.apache.kafka.common.network.SslTransportLayer.flush(SslTransportLayer.java:209)
at org.apache.kafka.common.network.SslTransportLayer.close(SslTransportLayer.java:172)
at org.apache.kafka.common.utils.Utils.closeAll(Utils.java:718)
at org.apache.kafka.common.network.KafkaChannel.close(KafkaChannel.java:61)
at org.apache.kafka.common.network.Selector.doClose(Selector.java:746)
at org.apache.kafka.common.network.Selector.close(Selector.java:734)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:532)
at org.apache.kafka.common.network.Selector.poll(Selector.java:424)
at kafka.network.Processor.poll(SocketServer.scala:628)
at kafka.network.Processor.run(SocketServer.scala:545)
at java.lang.Thread.run(Thread.java:748)
我可以从命令提示符发布到主题。
当我在KafkaPublish中使用SASL_SSL
时,Kerberos服务名称是必填字段。但是既然我们没有使用Kerberos,我应该为这个字段设置什么值?
当我设置动态属性sasl.mechanism和sasl.jaas.configuration时,我得到无效的配置参数警告。
我错过了任何配置吗?任何帮助将不胜感激。提前致谢。
自版本0_11以来,NiFi的PublishKafka仅支持SASL以及Kerberos以外的机制。
因此,如果您需要使用SASL Plain,请确保使用PublishKafka_0_11或PublishKafka_1_0。