Exposition de brokers Kafka sur deux interfaces réseaux

Une installation Big Data nécessite généralement d’avoir plusieurs interfaces réseaux, nous allons voir comment paramétrer Kafka pour écouter sur plus d’une.

La procédure décrite dans cette article a été testée sur CDH 5.7.1 avec Kafka 2.0.1.5 installé en parcels.

L’un des clusters sur lesquels nous travaillons a la configuration réseau suivante :

• Un réseau “data” exposant les noeuds clients, Kafka et les noeuds master
• Un réseau “interne” dédié au cluster pour nos workers

Nous utilisons Kafka pour l’ingestion de données et pour dispatcher des données enrichies vers un autre système exposant des interfaces utilisateurs avec :

• Un job Spark Streaming consumant des topics Kafka depuis YARN (réseau “interne”)
• Le système front consumant des topics Kafka depuis l’extérieur (réseau “data”)

Ainsi nous avons besoin de pouvoir atteindre Kafka sur deux réseaux différents. Pour que cela fonctionne il faut que les noeuds Kafka aient le même hostname sur les deux réseaux et que nous appliquions la configuration suivante sur chacun des brokers Kafka dans la section kafka.properties safety valve :

listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://<hostname>:9092

Et c’est tout !

NB : avec cette configuration, Kafka écoute sur toutes les interfaces et non plus celle par défaut. Théoriquement, la configuration suivante permet d’indiquer des IPs spécifiques :

listeners=PLAINTEXT://<ip1>:9092,PLAINTEXT://<ip2>:9092
advertised.listeners=PLAINTEXT://<hostname>:9092

Cependant l’exception suivante est lancée au démarrage :

java.lang.IllegalArgumentException: requirement failed: Each listener must have a different port
  at scala.Predef$.require(Predef.scala:219)
  at kafka.server.KafkaConfig.validateUniquePortAndProtocol(KafkaConfig.scala:905)
  at kafka.server.KafkaConfig.getListeners(KafkaConfig.scala:913)
  at kafka.server.KafkaConfig.<init>(KafkaConfig.scala:866)
  at kafka.server.KafkaConfig$.fromProps(KafkaConfig.scala:698)
  at kafka.server.KafkaConfig$.fromProps(KafkaConfig.scala:695)
  at kafka.server.KafkaServerStartable$.fromProps(KafkaServerStartable.scala:28)
  at kafka.Kafka$.main(Kafka.scala:58)
  at com.cloudera.kafka.wrap.Kafka$.main(Kafka.scala:76)
  at com.cloudera.kafka.wrap.Kafka.main(Kafka.scala)

et des variations de ”Each listener must have a different protocol” en changeant les ports.