src/rdkafka_broker.c
void rd_kafka_broker_update (rd_kafka_t *rk, rd_kafka_secproto_t proto,
                             const struct rd_kafka_metadata_broker *mdb) {
	rd_kafka_broker_t *rkb;
        char nodename[RD_KAFKA_NODENAME_SIZE];
        int needs_update = 0;

        rd_kafka_mk_nodename(nodename, sizeof(nodename), mdb->host, mdb->port);

	rd_kafka_wrlock(rk);
	if (unlikely(rd_kafka_terminating(rk))) {
		/* Dont update metadata while terminating, do this
		 * after acquiring lock for proper synchronisation */
		rd_kafka_wrunlock(rk);
		return;
	}

	if ((rkb = rd_kafka_broker_find_by_nodeid(rk, mdb->id))) {
                /* Broker matched by nodeid, see if we need to update
                 * the hostname. */
                if (strcmp(rkb->rkb_nodename, nodename))
                        needs_update = 1;
        } else if ((rkb = rd_kafka_broker_find(rk, proto,
					       mdb->host, mdb->port))) {
                /* Broker matched by hostname (but not by nodeid),
                 * update the nodeid. */
                needs_update = 1;

        } else {
		rd_kafka_broker_add(rk, RD_KAFKA_LEARNED,
				    proto, mdb->host, mdb->port, mdb->id);
	}

	rd_kafka_wrunlock(rk);

        if (rkb) {
                /* Existing broker */
                if (needs_update) {
                        rd_kafka_op_t *rko;

                        rko = rd_kafka_op_new(RD_KAFKA_OP_NODE_UPDATE);
                        strncpy(rko->rko_u.node.nodename, nodename,
				sizeof(rko->rko_u.node.nodename)-1);
                        rko->rko_u.node.nodeid   = mdb->id;
                        rd_kafka_q_enq(rkb->rkb_ops, rko);
                }
                rd_kafka_broker_destroy(rkb);
        }
}
/**
 * Adds a new broker or updates an existing one.
 *
 */
Adds a new broker or updates an existing one.
Adds a new broker or updates an existing one .
void rd_kafka_broker_update (rd_kafka_t *rk, rd_kafka_secproto_t proto,
                             const struct rd_kafka_metadata_broker *mdb) {
	rd_kafka_broker_t *rkb;
        char nodename[RD_KAFKA_NODENAME_SIZE];
        int needs_update = 0;
        rd_kafka_mk_nodename(nodename, sizeof(nodename), mdb->host, mdb->port);
	rd_kafka_wrlock(rk);
	if (unlikely(rd_kafka_terminating(rk))) {
		rd_kafka_wrunlock(rk);
		return;
	}
	if ((rkb = rd_kafka_broker_find_by_nodeid(rk, mdb->id))) {
                if (strcmp(rkb->rkb_nodename, nodename))
                        needs_update = 1;
        } else if ((rkb = rd_kafka_broker_find(rk, proto,
					       mdb->host, mdb->port))) {
                needs_update = 1;
        } else {
		rd_kafka_broker_add(rk, RD_KAFKA_LEARNED,
				    proto, mdb->host, mdb->port, mdb->id);
	}
	rd_kafka_wrunlock(rk);
        if (rkb) {
                if (needs_update) {
                        rd_kafka_op_t *rko;
                        rko = rd_kafka_op_new(RD_KAFKA_OP_NODE_UPDATE);
                        strncpy(rko->rko_u.node.nodename, nodename,
				sizeof(rko->rko_u.node.nodename)-1);
                        rko->rko_u.node.nodeid   = mdb->id;
                        rd_kafka_q_enq(rkb->rkb_ops, rko);
                }
                rd_kafka_broker_destroy(rkb);
        }
}
void rd_kafka_broker_update ( rd_kafka_t * rk , rd_kafka_secproto_t proto , const struct rd_kafka_metadata_broker * mdb ) { rd_kafka_broker_t * rkb ; char nodename [ RD_KAFKA_NODENAME_SIZE ] ; int needs_update = 0 ; rd_kafka_mk_nodename ( nodename , sizeof ( nodename ) , mdb -> host , mdb -> port ) ; rd_kafka_wrlock ( rk ) ; if ( unlikely ( rd_kafka_terminating ( rk ) ) ) { rd_kafka_wrunlock ( rk ) ; return ; } if ( ( rkb = rd_kafka_broker_find_by_nodeid ( rk , mdb -> id ) ) ) { if ( strcmp ( rkb -> rkb_nodename , nodename ) ) needs_update = 1 ; } else if ( ( rkb = rd_kafka_broker_find ( rk , proto , mdb -> host , mdb -> port ) ) ) { needs_update = 1 ; } else { rd_kafka_broker_add ( rk , RD_KAFKA_LEARNED , proto , mdb -> host , mdb -> port , mdb -> id ) ; } rd_kafka_wrunlock ( rk ) ; if ( rkb ) { if ( needs_update ) { rd_kafka_op_t * rko ; rko = rd_kafka_op_new ( RD_KAFKA_OP_NODE_UPDATE ) ; strncpy ( rko -> rko_u . node . nodename , nodename , sizeof ( rko -> rko_u . node . nodename ) - 1 ) ; rko -> rko_u . node . nodeid = mdb -> id ; rd_kafka_q_enq ( rkb -> rkb_ops , rko ) ; } rd_kafka_broker_destroy ( rkb ) ; } }
Adds a new broker or updates an existing one.
Adds a new broker or updates an existing one .
/* Dont update metadata while terminating, do this
		 * after acquiring lock for proper synchronisation */ /* Broker matched by nodeid, see if we need to update
                 * the hostname. */ /* Broker matched by hostname (but not by nodeid),
                 * update the nodeid. */ /* Existing broker */