src/rdkafka_broker.c
static int rd_kafka_broker_connect (rd_kafka_broker_t *rkb) {
	const rd_sockaddr_inx_t *sinx;
	char errstr[512];
        char nodename[RD_KAFKA_NODENAME_SIZE];

	rd_rkb_dbg(rkb, BROKER, "CONNECT",
		"broker in state %s connecting",
		rd_kafka_broker_state_names[rkb->rkb_state]);

        rd_atomic32_add(&rkb->rkb_c.connects, 1);

        rd_kafka_broker_lock(rkb);
        strncpy(nodename, rkb->rkb_nodename, sizeof(nodename));
        rkb->rkb_connect_epoch = rkb->rkb_nodename_epoch;
        /* Logical brokers might not have a hostname set, in which case
         * we should not try to connect. */
        if (*nodename)
                rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_CONNECT);
        rd_kafka_broker_unlock(rkb);

        if (!*nodename) {
                rd_rkb_dbg(rkb, BROKER, "CONNECT",
                           "broker has no address yet: postponing connect");
                return 0;
        }

        rd_kafka_broker_update_reconnect_backoff(rkb, &rkb->rkb_rk->rk_conf,
                                                 rd_clock());

        if (rd_kafka_broker_resolve(rkb, nodename) == -1)
                return -1;

	sinx = rd_sockaddr_list_next(rkb->rkb_rsal);

	rd_kafka_assert(rkb->rkb_rk, !rkb->rkb_transport);

	if (!(rkb->rkb_transport = rd_kafka_transport_connect(rkb, sinx,
		errstr, sizeof(errstr)))) {
		/* Avoid duplicate log messages */
		if (rkb->rkb_err.err == errno)
			rd_kafka_broker_fail(rkb, LOG_DEBUG,
                                             RD_KAFKA_RESP_ERR__FAIL, NULL);
		else
			rd_kafka_broker_fail(rkb, LOG_ERR,
                                             RD_KAFKA_RESP_ERR__TRANSPORT,
					     "%s", errstr);
		return -1;
	}

	return 0;
}
/**
 * @brief Initiate asynchronous connection attempt to the next address
 *        in the broker's address list.
 *        While the connect is asynchronous and its IO served in the
 *        CONNECT state, the initial name resolve is blocking.
 *
 * @returns -1 on error, 0 if broker does not have a hostname, or 1
 *          if the connection is now in progress.
 */
@brief Initiate asynchronous connection attempt to the next address
in the broker's address list.
While the connect is asynchronous and its IO served in the
CONNECT state, the initial name resolve is blocking.

@returns -1 on error, 0 if broker does not have a hostname, or 1
if the connection is now in progress.
@brief Initiate asynchronous connection attempt to the next address in the broker ' s address list . While the connect is asynchronous and its IO served in the CONNECT state the initial name resolve is blocking . @returns - 1 on error 0 if broker does not have a hostname or 1 if the connection is now in progress .
static int rd_kafka_broker_connect (rd_kafka_broker_t *rkb) {
	const rd_sockaddr_inx_t *sinx;
	char errstr[512];
        char nodename[RD_KAFKA_NODENAME_SIZE];
	rd_rkb_dbg(rkb, BROKER, "CONNECT",
		"broker in state %s connecting",
		rd_kafka_broker_state_names[rkb->rkb_state]);
        rd_atomic32_add(&rkb->rkb_c.connects, 1);
        rd_kafka_broker_lock(rkb);
        strncpy(nodename, rkb->rkb_nodename, sizeof(nodename));
        rkb->rkb_connect_epoch = rkb->rkb_nodename_epoch;
        if (*nodename)
                rd_kafka_broker_set_state(rkb, RD_KAFKA_BROKER_STATE_CONNECT);
        rd_kafka_broker_unlock(rkb);
        if (!*nodename) {
                rd_rkb_dbg(rkb, BROKER, "CONNECT",
                           "broker has no address yet: postponing connect");
                return 0;
        }
        rd_kafka_broker_update_reconnect_backoff(rkb, &rkb->rkb_rk->rk_conf,
                                                 rd_clock());
        if (rd_kafka_broker_resolve(rkb, nodename) == -1)
                return -1;
	sinx = rd_sockaddr_list_next(rkb->rkb_rsal);
	rd_kafka_assert(rkb->rkb_rk, !rkb->rkb_transport);
	if (!(rkb->rkb_transport = rd_kafka_transport_connect(rkb, sinx,
		errstr, sizeof(errstr)))) {
		if (rkb->rkb_err.err == errno)
			rd_kafka_broker_fail(rkb, LOG_DEBUG,
                                             RD_KAFKA_RESP_ERR__FAIL, NULL);
		else
			rd_kafka_broker_fail(rkb, LOG_ERR,
                                             RD_KAFKA_RESP_ERR__TRANSPORT,
					     "%s", errstr);
		return -1;
	}
	return 0;
}
static int rd_kafka_broker_connect ( rd_kafka_broker_t * rkb ) { const rd_sockaddr_inx_t * sinx ; char errstr [ 512 ] ; char nodename [ RD_KAFKA_NODENAME_SIZE ] ; rd_rkb_dbg ( rkb , BROKER , " " , " " , rd_kafka_broker_state_names [ rkb -> rkb_state ] ) ; rd_atomic32_add ( & rkb -> rkb_c . connects , 1 ) ; rd_kafka_broker_lock ( rkb ) ; strncpy ( nodename , rkb -> rkb_nodename , sizeof ( nodename ) ) ; rkb -> rkb_connect_epoch = rkb -> rkb_nodename_epoch ; if ( * nodename ) rd_kafka_broker_set_state ( rkb , RD_KAFKA_BROKER_STATE_CONNECT ) ; rd_kafka_broker_unlock ( rkb ) ; if ( ! * nodename ) { rd_rkb_dbg ( rkb , BROKER , " " , " " ) ; return 0 ; } rd_kafka_broker_update_reconnect_backoff ( rkb , & rkb -> rkb_rk -> rk_conf , rd_clock ( ) ) ; if ( rd_kafka_broker_resolve ( rkb , nodename ) == -1 ) return -1 ; sinx = rd_sockaddr_list_next ( rkb -> rkb_rsal ) ; rd_kafka_assert ( rkb -> rkb_rk , ! rkb -> rkb_transport ) ; if ( ! ( rkb -> rkb_transport = rd_kafka_transport_connect ( rkb , sinx , errstr , sizeof ( errstr ) ) ) ) { if ( rkb -> rkb_err . err == errno ) rd_kafka_broker_fail ( rkb , LOG_DEBUG , RD_KAFKA_RESP_ERR__FAIL , NULL ) ; else rd_kafka_broker_fail ( rkb , LOG_ERR , RD_KAFKA_RESP_ERR__TRANSPORT , " " , errstr ) ; return -1 ; } return 0 ; }
@brief Initiate asynchronous connection attempt to the next address
        in the broker's address list.
@brief Initiate asynchronous connection attempt to the next address in the broker ' s address list .
/* Logical brokers might not have a hostname set, in which case
         * we should not try to connect. */ /* Avoid duplicate log messages */