src/rdkafka_broker.c
static void rd_kafka_broker_connect_auth (rd_kafka_broker_t *rkb) {

	if ((rkb->rkb_proto == RD_KAFKA_PROTO_SASL_PLAINTEXT ||
	     rkb->rkb_proto == RD_KAFKA_PROTO_SASL_SSL)) {

		rd_rkb_dbg(rkb, SECURITY | RD_KAFKA_DBG_BROKER, "AUTH",
			   "Auth in state %s (handshake %ssupported)",
			   rd_kafka_broker_state_names[rkb->rkb_state],
			   (rkb->rkb_features&RD_KAFKA_FEATURE_SASL_HANDSHAKE)
			   ? "" : "not ");

		/* Broker >= 0.10.0: send request to select mechanism */
		if (rkb->rkb_state != RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE &&
		    (rkb->rkb_features & RD_KAFKA_FEATURE_SASL_HANDSHAKE)) {

			rd_kafka_broker_lock(rkb);
			rd_kafka_broker_set_state(
				rkb, RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE);
			rd_kafka_broker_unlock(rkb);

			rd_kafka_SaslHandshakeRequest(
				rkb, rkb->rkb_rk->rk_conf.sasl.mechanisms,
				RD_KAFKA_NO_REPLYQ,
				rd_kafka_broker_handle_SaslHandshake,
				NULL);
		} else {
			/* Either Handshake succeeded (protocol selected)
			 * or Handshakes were not supported.
			 * In both cases continue with authentication. */
			char sasl_errstr[512];

			rd_kafka_broker_lock(rkb);
			rd_kafka_broker_set_state(rkb,
						  RD_KAFKA_BROKER_STATE_AUTH);
			rd_kafka_broker_unlock(rkb);

			if (rd_kafka_sasl_client_new(
				    rkb->rkb_transport, sasl_errstr,
				    sizeof(sasl_errstr)) == -1) {
				errno = EINVAL;
				rd_kafka_broker_fail(
					rkb, LOG_ERR,
					RD_KAFKA_RESP_ERR__AUTHENTICATION,
					"Failed to initialize "
					"SASL authentication: %s",
					sasl_errstr);
				return;
			}

			/* Enter non-Kafka-protocol-framed SASL communication
			 * state handled in rdkafka_sasl.c */
			rd_kafka_broker_lock(rkb);
			rd_kafka_broker_set_state(rkb,
						  RD_KAFKA_BROKER_STATE_AUTH);
			rd_kafka_broker_unlock(rkb);
		}

		return;
	}

	/* No authentication required. */
	rd_kafka_broker_connect_up(rkb);
}
/**
 * @brief Transition state to:
 *        - AUTH_HANDSHAKE (if SASL is configured and handshakes supported)
 *        - AUTH (if SASL is configured but no handshake is required or
 *                not supported, or has already taken place.)
 *        - UP (if SASL is not configured)
 */
@brief Transition state to:
- AUTH_HANDSHAKE (if SASL is configured and handshakes supported)
- AUTH (if SASL is configured but no handshake is required or
not supported, or has already taken place.)
- UP (if SASL is not configured)
@brief Transition state to : - AUTH_HANDSHAKE ( if SASL is configured and handshakes supported ) - AUTH ( if SASL is configured but no handshake is required or not supported or has already taken place . ) - UP ( if SASL is not configured )
static void rd_kafka_broker_connect_auth (rd_kafka_broker_t *rkb) {
	if ((rkb->rkb_proto == RD_KAFKA_PROTO_SASL_PLAINTEXT ||
	     rkb->rkb_proto == RD_KAFKA_PROTO_SASL_SSL)) {
		rd_rkb_dbg(rkb, SECURITY | RD_KAFKA_DBG_BROKER, "AUTH",
			   "Auth in state %s (handshake %ssupported)",
			   rd_kafka_broker_state_names[rkb->rkb_state],
			   (rkb->rkb_features&RD_KAFKA_FEATURE_SASL_HANDSHAKE)
			   ? "" : "not ");
		if (rkb->rkb_state != RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE &&
		    (rkb->rkb_features & RD_KAFKA_FEATURE_SASL_HANDSHAKE)) {
			rd_kafka_broker_lock(rkb);
			rd_kafka_broker_set_state(
				rkb, RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE);
			rd_kafka_broker_unlock(rkb);
			rd_kafka_SaslHandshakeRequest(
				rkb, rkb->rkb_rk->rk_conf.sasl.mechanisms,
				RD_KAFKA_NO_REPLYQ,
				rd_kafka_broker_handle_SaslHandshake,
				NULL);
		} else {
			char sasl_errstr[512];
			rd_kafka_broker_lock(rkb);
			rd_kafka_broker_set_state(rkb,
						  RD_KAFKA_BROKER_STATE_AUTH);
			rd_kafka_broker_unlock(rkb);
			if (rd_kafka_sasl_client_new(
				    rkb->rkb_transport, sasl_errstr,
				    sizeof(sasl_errstr)) == -1) {
				errno = EINVAL;
				rd_kafka_broker_fail(
					rkb, LOG_ERR,
					RD_KAFKA_RESP_ERR__AUTHENTICATION,
					"Failed to initialize "
					"SASL authentication: %s",
					sasl_errstr);
				return;
			}
			rd_kafka_broker_lock(rkb);
			rd_kafka_broker_set_state(rkb,
						  RD_KAFKA_BROKER_STATE_AUTH);
			rd_kafka_broker_unlock(rkb);
		}
		return;
	}
	rd_kafka_broker_connect_up(rkb);
}
static void rd_kafka_broker_connect_auth ( rd_kafka_broker_t * rkb ) { if ( ( rkb -> rkb_proto == RD_KAFKA_PROTO_SASL_PLAINTEXT || rkb -> rkb_proto == RD_KAFKA_PROTO_SASL_SSL ) ) { rd_rkb_dbg ( rkb , SECURITY | RD_KAFKA_DBG_BROKER , " " , " " , rd_kafka_broker_state_names [ rkb -> rkb_state ] , ( rkb -> rkb_features & RD_KAFKA_FEATURE_SASL_HANDSHAKE ) ? " " : " " ) ; if ( rkb -> rkb_state != RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE && ( rkb -> rkb_features & RD_KAFKA_FEATURE_SASL_HANDSHAKE ) ) { rd_kafka_broker_lock ( rkb ) ; rd_kafka_broker_set_state ( rkb , RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE ) ; rd_kafka_broker_unlock ( rkb ) ; rd_kafka_SaslHandshakeRequest ( rkb , rkb -> rkb_rk -> rk_conf . sasl . mechanisms , RD_KAFKA_NO_REPLYQ , rd_kafka_broker_handle_SaslHandshake , NULL ) ; } else { char sasl_errstr [ 512 ] ; rd_kafka_broker_lock ( rkb ) ; rd_kafka_broker_set_state ( rkb , RD_KAFKA_BROKER_STATE_AUTH ) ; rd_kafka_broker_unlock ( rkb ) ; if ( rd_kafka_sasl_client_new ( rkb -> rkb_transport , sasl_errstr , sizeof ( sasl_errstr ) ) == -1 ) { errno = EINVAL ; rd_kafka_broker_fail ( rkb , LOG_ERR , RD_KAFKA_RESP_ERR__AUTHENTICATION , " " " " , sasl_errstr ) ; return ; } rd_kafka_broker_lock ( rkb ) ; rd_kafka_broker_set_state ( rkb , RD_KAFKA_BROKER_STATE_AUTH ) ; rd_kafka_broker_unlock ( rkb ) ; } return ; } rd_kafka_broker_connect_up ( rkb ) ; }
@brief Transition state to:
        - AUTH_HANDSHAKE (if SASL is configured and handshakes supported)
        - AUTH (if SASL is configured but no handshake is required or
                not supported, or has already taken place.)
@brief Transition state to : - AUTH_HANDSHAKE ( if SASL is configured and handshakes supported ) - AUTH ( if SASL is configured but no handshake is required or not supported or has already taken place . )
/* Broker >= 0.10.0: send request to select mechanism */ /* Either Handshake succeeded (protocol selected)
			 * or Handshakes were not supported.
			 * In both cases continue with authentication. */ /* Enter non-Kafka-protocol-framed SASL communication
			 * state handled in rdkafka_sasl.c */ /* No authentication required. */