apache-kafka - 需要帮助来理解 ansible j2

Question

我需要帮助来理解下面的代码kafka.properties.j2。

{% for server in kafka_group %}
{% if server.host is defined %}
{% if server.host == inventory_hostname %}
broker.id={{ server.id }}
advertised.listeners=PLAINTEXT://{{ hostvars.get(server).ansible_host }}:{{ kafka_plaintext_port }}
listeners=PLAINTEXT://{{ hostvars.get(server).ansible_host }}:{{ kafka_plaintext_port }}
{% endif %}
{% else %}
{% if server == inventory_hostname %}
broker.id={{ loop.index }}
advertised.listeners=PLAINTEXT://{{ hostvars.get(server).ansible_host }}:{{ kafka_plaintext_port }}
listeners=PLAINTEXT://{{ hostvars.get(server).ansible_host }}:{{ kafka_plaintext_port }}
{% endif %}
{% endif %}
{% endfor %}

通过 Ansible-playbook 运行上述代码后，它配置了kafka.properties如下所示的文件。

broker.id=1
advertised.listeners=PLAINTEXT://54.154.254.129:3997
listeners=PLAINTEXT://54.154.254.129:3997

在下面的 .ini 文件中是 kafka 组的条目。

testdev01 ansible_host=54.154.254.129 ansible_user=ec2-user

[kafka]
testdev01

kafka_ip =172.40.0.20

我面临的问题是kafka.properties3997 绑定到公共 ip 并且 kafka 服务无法启动。将其更改为 kafka_ip (172.40.0.20:3997) 后服务正确启动，因此我需要了解。
第一if server.host部分是真的还是if server。和上面
有什么区别。 它如何选择为1，因此进入文件。 为什么这个变量选择而不是私有IP。if server.host ==if server ==
server.idbroker.id=1
hostvars.get(server).ansible_host54.154.254.129kafka_ip

列出 Kafka 主题时，剧本出现以下错误。

fatal: [testdev01]: FAILED! => {"changed": false, "cmd": ["/opt/kafka/kafka_2.12-2.2.2/bin/kafka-topics.sh", "--zookeeper", "54.154.254.129:5181", "--list"], "delta": "0:00:31.730705", "end": "2020-11-28 12:29:52.207646", "msg": "non-zero return code", "rc": 1, "start": "2020-11-28 12:29:20.476941", "stderr": "[2020-11-28 12:29:51,779] WARN Client session timed out, have not heard from server in 30007ms for sessionid 0x0 (org.apache.zookeeper.ClientCnxn)\nException in thread \"main\" kafka.zookeeper.ZooKeeperClientTimeoutException: Timed out waiting for connection while in state: CONNECTING\n\tat kafka.zookeeper.ZooKeeperClient.$anonfun$waitUntilConnected$3(ZooKeeperClient.scala:242)\n\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n\tat kafka.utils.CoreUtils$.inLock(CoreUtils.scala:251)\n\tat kafka.zookeeper.ZooKeeperClient.waitUntilConnected(ZooKeeperClient.scala:238)\n\tat kafka.zookeeper.ZooKeeperClient.<init>(ZooKeeperClient.scala:96)\n\tat kafka.zk.KafkaZkClient$.apply(KafkaZkClient.scala:1825)\n\tat kafka.admin.TopicCommand$ZookeeperTopicService$.apply(TopicCommand.scala:262)\n\tat kafka.admin.TopicCommand$.main(TopicCommand.scala:53)\n\tat kafka.admin.TopicCommand.main(TopicCommand.scala)", "stderr_lines": ["[2020-11-28 12:29:51,779] WARN Client session timed out, have not heard from server in 30007ms for sessionid 0x0 (org.apache.zookeeper.ClientCnxn)", "Exception in thread \"main\" kafka.zookeeper.ZooKeeperClientTimeoutException: Timed out waiting for connection while in state: CONNECTING", "\tat kafka.zookeeper.ZooKeeperClient.$anonfun$waitUntilConnected$3(ZooKeeperClient.scala:242)", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)", "\tat kafka.utils.CoreUtils$.inLock(CoreUtils.scala:251)", "\tat kafka.zookeeper.ZooKeeperClient.waitUntilConnected(ZooKeeperClient.scala:238)", "\tat kafka.zookeeper.ZooKeeperClient.<init>(ZooKeeperClient.scala:96)", "\tat kafka.zk.KafkaZkClient$.apply(KafkaZkClient.scala:1825)", "\tat kafka.admin.TopicCommand$ZookeeperTopicService$.apply(TopicCommand.scala:262)", "\tat kafka.admin.TopicCommand$.main(TopicCommand.scala:53)", "\tat kafka.admin.TopicCommand.main(TopicCommand.scala)"], "stdout": "", "stdout_lines": []}

kafka.log错误

[2020-11-28 12:29:20,687] ERROR [KafkaServer id=1] Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.
KafkaServer)
org.apache.kafka.common.KafkaException: Socket server failed to bind to 54.154.254.129:3997: Cannot assign requested address.
        at kafka.network.Acceptor.openServerSocket(SocketServer.scala:573)
        at kafka.network.Acceptor.<init>(SocketServer.scala:451)
        at kafka.network.SocketServer.createAcceptor(SocketServer.scala:245)
        at kafka.network.SocketServer.$anonfun$createDataPlaneAcceptorsAndProcessors$1(SocketServer.scala:215)
        at kafka.network.SocketServer.$anonfun$createDataPlaneAcceptorsAndProcessors$1$adapted(SocketServer.scala:214)
        at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
        at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
        at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
        at kafka.network.SocketServer.createDataPlaneAcceptorsAndProcessors(SocketServer.scala:214)
        at kafka.network.SocketServer.startup(SocketServer.scala:114)
        at kafka.server.KafkaServer.startup(KafkaServer.scala:253)
        at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:38)
        at kafka.Kafka$.main(Kafka.scala:75)
        at kafka.Kafka.main(Kafka.scala)
Caused by: java.net.BindException: Cannot assign requested address
        at sun.nio.ch.Net.bind0(Native Method)
at sun.nio.ch.Net.bind(Net.java:433)
        at sun.nio.ch.Net.bind(Net.java:425)
        at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:223)
        at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:74)
        at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:67)
        at kafka.network.Acceptor.openServerSocket(SocketServer.scala:569)
        ... 13 more

[2020-11-28 12:29:20,690] INFO [KafkaServer id=1] shutting down (kafka.server.KafkaServer)
[2020-11-28 12:29:20,692] INFO [SocketServer brokerId=1] Stopping socket server request processors (kafka.network.SocketServer)
[2020-11-28 12:29:20,695] INFO [SocketServer brokerId=1] Stopped socket server request processors (kafka.network.SocketServer)
[2020-11-28 12:29:20,699] INFO Shutting down. (kafka.log.LogManager)
[2020-11-28 12:29:20,732] INFO Shutdown complete. (kafka.log.LogManager)

标签： apache-kafkaansible