Extract authentication URL and rename internal

function and enums

Author: emasab

src/rdhttp.c

@@ -367,14 +367,32 @@ const char *rd_http_req_get_content_type(rd_http_req_t *hreq) {
 
 /**
  * @brief Perform a blocking HTTP(S) request to \p url.
+ *        Retries the request \p retries times with linear backoff.
+ *        Interval of \p retry_ms milliseconds is used between retries.
  *
- * Returns the response (even if there's a HTTP error code returned)
- * in \p *rbufp.
+ * @param url The URL to perform the request to.
+ * @param headers_array Array of HTTP(S) headers to set, each element
+ *                      is a string in the form "key: value"
+ * @param headers_array_cnt Number of elements in \p headers_array.
+ * @param timeout_s Timeout in seconds for the request, 0 means default
+ *                  `rd_http_req_init()` timeout.
+ * @param retries Number of retries to perform on failure.
+ * @param retry_ms Milliseconds to wait between retries.
+ * @param rbufp (out) Pointer to a buffer that will be filled with the response.
+ * @param content_type (out, optional) Pointer to a string that will be filled
+ * with the content type of the response, if not NULL.
+ * @param response_code (out, optional) Pointer to an integer that will be
+ * filled with the HTTP response code, if not NULL.
  *
- * Returns NULL on success (HTTP response code < 400), or an error
- * object on transport or HTTP error - this error object must be destroyed
- * by calling rd_http_error_destroy(). In case of HTTP error the \p *rbufp
- * may be filled with the error response.
+ * @return Returns NULL on success (HTTP response code < 400), or an error
+ * object on transport or HTTP error.
+ *
+ * @remark Returned error object, when non-NULL, must be destroyed
+ *         by calling rd_http_error_destroy().
+ *
+ * @locality Any thread.
+ * @locks None.
+ * @locks_acquired None.
  */
 rd_http_error_t *rd_http_get(rd_kafka_t *rk,
                              const char *url,
@@ -409,6 +427,8 @@ rd_http_error_t *rd_http_get(rd_kafka_t *rk,
                         headers = curl_slist_append(headers, header);
         }
         curl_easy_setopt(hreq.hreq_curl, CURLOPT_HTTPHEADER, headers);
+        if (timeout_s > 0)
+                curl_easy_setopt(hreq.hreq_curl, CURLOPT_TIMEOUT, timeout_s);
 
         for (i = 0; i <= retries; i++) {
                 if (rd_kafka_terminating(rk)) {

src/rdkafka.c

@@ -2461,10 +2461,10 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,
                 rk->rk_conf.sasl.oauthbearer.builtin_token_refresh_cb = rd_true;
 
                 if (rk->rk_conf.sasl.oauthbearer.metadata_authentication.type ==
-                    RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_AZURE) {
+                    RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_AZURE_IMDS) {
                         rd_kafka_conf_set_oauthbearer_token_refresh_cb(
                             &rk->rk_conf,
-                            rd_kafka_oidc_token_metadata_azure_refresh_cb);
+                            rd_kafka_oidc_token_metadata_azure_imds_refresh_cb);
                 } else if (
                     rk->rk_conf.sasl.oauthbearer.grant_type ==
                     RD_KAFKA_SASL_OAUTHBEARER_GRANT_TYPE_CLIENT_CREDENTIALS) {

src/rdkafka_conf.c

@@ -1222,10 +1222,11 @@ static const struct rd_kafka_property rd_kafka_properties[] = {
         "through `sasl.oauthbearer.config`.",
         _UNSUPPORTED_OIDC,
         .vdef = RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_NONE,
-        .s2i  = {{RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_NONE,
-                  "none"},
-                 {RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_AZURE,
-                  "azure_imds"}},
+        .s2i =
+            {{RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_NONE,
+              "none"},
+             {RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_AZURE_IMDS,
+              "azure_imds"}},
     },
 
     /* Plugins */
@@ -4066,10 +4067,10 @@ const char *rd_kafka_conf_finalize_oauthbearer_oidc(rd_kafka_conf_t *conf) {
                        "mutually exclusive";
 
         if (conf->sasl.oauthbearer.metadata_authentication.type ==
-                RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_AZURE &&
+                RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_AZURE_IMDS &&
             !conf->sasl.oauthbearer.token_endpoint_url) {
                 conf->sasl.oauthbearer.token_endpoint_url =
-                    "http://169.254.169.254/metadata/identity/oauth2/token";
+                    RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_URL_AZURE_IMDS;
         }
 
         if (!conf->sasl.oauthbearer.token_endpoint_url) {

src/rdkafka_conf.h

@@ -166,9 +166,14 @@ typedef enum {
 
 typedef enum {
         RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_NONE,
-        RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_AZURE,
+        RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_TYPE_AZURE_IMDS,
 } rd_kafka_oauthbearer_metadata_authentication_type_t;
 
+
+#define RD_KAFKA_SASL_OAUTHBEARER_METADATA_AUTHENTICATION_URL_AZURE_IMDS       \
+        "http://169.254.169.254/metadata/identity/oauth2/token"
+
+
 typedef enum {
         RD_KAFKA_SSL_ENDPOINT_ID_NONE,
         RD_KAFKA_SSL_ENDPOINT_ID_HTTPS, /**< RFC2818 */

src/rdkafka_sasl_oauthbearer_oidc.c

@@ -1000,12 +1000,13 @@ void rd_kafka_oidc_token_client_credentials_refresh_cb(
 }
 
 /**
- * @brief Implementation of Oauth/OIDC token refresh callback function,
- *        will receive the JSON response after HTTP call to token provider,
- *        then extract the jwt from the JSON response, and forward it to
- *        the broker.
+ * @brief Implementation of Oauth/OIDC token refresh callback function
+ *        for Azure IMDS,
+ *        will receive the JSON response after HTTP(S) GET call to token
+ * provider, then extract the jwt from the JSON response, and forward it to the
+ * broker.
  */
-void rd_kafka_oidc_token_metadata_azure_refresh_cb(
+void rd_kafka_oidc_token_metadata_azure_imds_refresh_cb(
     rd_kafka_t *rk,
     const char *oauthbearer_config,
     void *opaque) {

src/rdkafka_sasl_oauthbearer_oidc.h

@@ -37,7 +37,7 @@ void rd_kafka_oidc_token_client_credentials_refresh_cb(
     const char *oauthbearer_config,
     void *opaque);
 
-void rd_kafka_oidc_token_metadata_azure_refresh_cb(
+void rd_kafka_oidc_token_metadata_azure_imds_refresh_cb(
     rd_kafka_t *rk,
     const char *oauthbearer_config,
     void *opaque);