Added orig_broker_name to metadata

..since orig_broker_id will be -1 for bootstrap brokers.

Author: edenhill

examples/rdkafka_example.c

@@ -168,9 +168,10 @@ static void metadata_print (const char *topic,
                             const struct rd_kafka_metadata *metadata) {
         int i, j, k;
 
-        printf("Metadata for %s (from broker %"PRId32"):\n",
+        printf("Metadata for %s (from broker %"PRId32": %s):\n",
                topic ? : "all topics",
-               metadata->orig_broker_id);
+               metadata->orig_broker_id,
+               metadata->orig_broker_name);
 
 
         /* Iterate brokers */

src/rdkafka.h

@@ -895,6 +895,7 @@ struct rd_kafka_metadata {
         struct rd_kafka_metadata_topic *topics;    /* Topics */
 
         int32_t     orig_broker_id; /* Broker originating this metadata */
+        char       *orig_broker_name; /* Name of originating broker */
 };
 
 

src/rdkafka_broker.c

@@ -797,14 +797,17 @@ rd_kafka_metadata_handle (rd_kafka_broker_t *rkb,
         int   msh_of  = 0;
         int   msh_size;
         struct rd_kafka_metadata *md = NULL;
+        int rkb_namelen = strlen(rkb->rkb_name)+1;
 
         /* We assume that the marshalled representation is
          * no more than 4 times larger than the wire representation. */
-        msh_size = sizeof(*md) + (size * 4);
+        msh_size = sizeof(*md) + rkb_namelen + (size * 4);
         msh_buf = malloc(msh_size);
 
         _MSH_ALLOC(md, sizeof(*md));
         md->orig_broker_id = rkb->rkb_nodeid;
+        _MSH_ALLOC(md->orig_broker_name, rkb_namelen);
+        memcpy(md->orig_broker_name, rkb->rkb_name, rkb_namelen);
 
 	/* Read Brokers */
 	_READ_I32A(md->broker_cnt);