Блог

Мне нужна помощь, чтобы понять приведенный ниже код для 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 group.

 testdev01 ansible_host=54.154.254.129 ansible_user=ec2-user

[kafka]
testdev01

kafka_ip =172.40.0.20
 

проблема, с которой я столкнулся, заключается в kafka.properties том, что 3997 получил привязку к общедоступному ip, и служба kafka не смогла запуститься. после изменения его на kafka_ip (172.40.0.20: 3997) служба запустилась правильно, поэтому мне нужно понять.
независимо от того, получила ли первая if server.host часть true или if server .
в чем разница между if server.host == и if server == выше.
как он был выбран server.id как 1, поэтому broker.id=1 попал в файл.
почему эта переменная hostvars.get(server).ansible_host выбирается 54.154.254.129 , а не kafka_ip т.е. частный ip.

playbook выдал ошибку ниже при перечислении тем 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: CONNECTINGntat kafka.zookeeper.ZooKeeperClient.$anonfun$waitUntilConnected$3(ZooKeeperClient.scala:242)ntat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)ntat kafka.utils.CoreUtils$.inLock(CoreUtils.scala:251)ntat kafka.zookeeper.ZooKeeperClient.waitUntilConnected(ZooKeeperClient.scala:238)ntat kafka.zookeeper.ZooKeeperClient.<init>(ZooKeeperClient.scala:96)ntat kafka.zk.KafkaZkClient$.apply(KafkaZkClient.scala:1825)ntat kafka.admin.TopicCommand$ZookeeperTopicService$.apply(TopicCommand.scala:262)ntat kafka.admin.TopicCommand$.main(TopicCommand.scala:53)ntat 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)