Dockerで立ち上げたKafkaを、ホスト環境でビルドしたJavaアプリケーションから叩こうとしたのだが、なぜかエラーになって叩けず。

java.lang.IllegalStateException: No entry found for connection 21729123512

よく調べてみると、advertised_listenerの設定を忘れていたことに気がついた。

advertised_listenerとは

KafkaのコアであるKafka brokerはListenerを持っている。クライアントからListenerに接続があると、そのListenerは自身がアクセスできるIPアドレスまたはホスト名を返し、それを使ってその後の処理が行われる。
しかし、Kafkaとアプリケーションが同じ環境にあれば全く問題ないが、Dockerを使っているとこれが問題になりうる。ListenerはDockerのインターナルネットワークの中にあるので、返ってくるホスト名もインターナルネットワークのものになってしまうのである。

現に、エラーが起きているKafkaに、ホスト環境からkafkacatで接続してみると以下のようになった。(kafkacatは、Kafka版netcatとして開発されたCLIであり、こういった接続確認などを行うときとても便利に使える)

$ kafkacat -b localhost:9092 -L
Metadata for all topics (from broker -1: localhost:9092/bootstrap):
 1 brokers:
  broker 0 at cd572956787b:9092
 29 topics:
...

broker 0 at cd572956787b:9092に注目。このcd572956787bとはKafkaのDockerコンテナのホスト名である。Dockerコンテナのインターナルネットワークのホスト名が返ってきてしまうために、ホスト環境ではこの名前解決ができず、その後のKafka brokerとの接続に失敗するわけだ。具体例のついたより詳しい説明が、この記事に書かれている。

rmoff.net

advertised_listenerを設定する

対応としては、Kafkaに外部ネットワークからアクセスできるような設定をすることが必要となる。server.properties内のadvertised.listenersがそれにあたる。これは外部ネットワークからアクセスする場合用に、Zookeeperに公開されるListenerの設定をするための項目になっている。
ここにadvertised.listeners=PLAINTEXT://<IP>:9092と設定すれば良い。なおこのIPはlocalhostでも192.168始まりのプライベートIPアドレスでも動作したが、kafka-dockerのREADMEによると、localhostだと複数のbrokerを動かすときに不具合が発生するとのことなので、プライベートIPアドレスを使う方が良いかもしれない。

landoop/fast-data-devを使っている場合はもっと簡単で、READMEにあるとおり、環境変数としてADV_HOST=<IP>を指定して起動させれば良い。

プライベートIPアドレスを指定して起動させ、再びkafkacatを叩いてみる。

$ kafkacat -b localhost:9092 -L
Metadata for all topics (from broker -1: localhost:9092/bootstrap):
 1 brokers:
  broker 0 at 192.168.58.109:9092
 29 topics:
...