Add a Kafka "metadata.broker.list" for each log writer filter.

   If a new log filter is added in bro and a specific kafka broker list is defined as:
       $config = table(["metadata.broker.list"] = "host:port")
   this will override the default broker list (only for this specific writer).

   If no specific "metadata.broker.list" for the writer is defined in the log filter, the
   default will be applied as in
      redef Kafka::kafka_conf = table(["metadata.broker.list"] = "host:port");

   Note: all other configuration settings will not be changed.

src/KafkaWriter.cc

@@ -169,6 +169,18 @@ bool KafkaWriter::DoInit(const WriterInfo& info, int num_fields, const threading
       }
     }
 
+     // Allow overriding of the kafka list of brokers via the Bro script constant 'metadata.broker.list'
+     // which can be applied when adding a new Bro log filter as $config = table(["metadata.broker.list"] = "host:port").
+     metadata_broker_list_override = GetConfigValue(info, "metadata.broker.list");
+     if ( !metadata_broker_list_override.empty() ) {
+        MsgThread::Info(Fmt("Overriding default metadata.broker.list with %s for writer %s.", metadata_broker_list_override.c_str(), info.path));
+       // apply overriding setting metadata.broker.list to kafka
+       if (RdKafka::Conf::CONF_OK != conf->set("metadata.broker.list", metadata_broker_list_override, err)) {
+           Error(Fmt("Failed to set '%s'='%s': %s", "metadata.broker.list", metadata_broker_list_override.c_str(), err.c_str()));
+           return false;
+       }
+     }
+
     if(is_debug) {
         string key("debug");
         string val(debug);

src/KafkaWriter.h

@@ -75,6 +75,7 @@ class KafkaWriter : public WriterBackend {
     map<string, string> kafka_conf;
     string topic_name;
     string topic_name_override;
+    string metadata_broker_list_override;
     threading::formatter::Formatter *formatter;
     RdKafka::Producer* producer;
     RdKafka::Topic* topic;