fix(aws): force credential refresh in provider refresh functions

- Add force refresh logic to EC2, STS, and EKS credential providers

- Set next_refresh to 0 in refresh functions to ensure immediate credential update

- Fixes MSK IAM authentication failures after ~1 hour due to stale credentials

- Aligns with AWS SDK behavior where refresh() means force refresh

This resolves the issue where OAuth token refresh (every ~15 minutes) would

not actually refresh AWS credentials until next_refresh time was reached

(typically 1 hour later), causing MSK connection failures with 'Access denied'

errors.

The fix ensures that every OAuth callback will fetch fresh credentials from

AWS, matching the behavior of official AWS SDKs (Python, Java).

Signed-off-by: Arbin <[email protected]>

Author: kalavt

src/aws/flb_aws_credentials_ec2.c

@@ -130,6 +130,10 @@ int refresh_fn_ec2(struct flb_aws_provider *provider) {
     int ret = -1;
 
     flb_debug("[aws_credentials] Refresh called on the EC2 IMDS provider");
+    
+    /* Force credential refresh by marking as expired */
+    implementation->next_refresh = 0;
+    
     if (try_lock_provider(provider)) {
         ret = get_creds_ec2(implementation);
         unlock_provider(provider);

src/aws/flb_aws_credentials_profile.c

@@ -663,8 +663,7 @@ static int get_shared_credentials(char* credentials_path,
 
     if (flb_read_file(credentials_path, &buf, &size) < 0) {
         if (errno == ENOENT) {
-            AWS_CREDS_ERROR_OR_DEBUG(debug_only, "Shared credentials file %s does not exist",
-                                     credentials_path);
+            AWS_CREDS_DEBUG("Shared credentials file %s does not exist", credentials_path);
         } else {
             flb_errno();
             AWS_CREDS_ERROR_OR_DEBUG(debug_only, "Could not read shared credentials file %s",

src/aws/flb_aws_credentials_sts.c

@@ -175,6 +175,9 @@ int refresh_fn_sts(struct flb_aws_provider *provider) {
     struct flb_aws_provider_sts *implementation = provider->implementation;
 
     flb_debug("[aws_credentials] Refresh called on the STS provider");
+    
+    /* Force credential refresh by marking as expired */
+    implementation->next_refresh = 0;
 
     if (try_lock_provider(provider)) {
         ret = sts_assume_role_request(implementation->sts_client,
@@ -480,6 +483,10 @@ int refresh_fn_eks(struct flb_aws_provider *provider) {
     struct flb_aws_provider_eks *implementation = provider->implementation;
 
     flb_debug("[aws_credentials] Refresh called on the EKS provider");
+    
+    /* Force credential refresh by marking as expired */
+    implementation->next_refresh = 0;
+    
     if (try_lock_provider(provider)) {
         ret = assume_with_web_identity(implementation);
         unlock_provider(provider);

src/aws/flb_aws_msk_iam.c

@@ -216,17 +216,17 @@ static flb_sds_t build_msk_iam_payload(struct flb_aws_msk_iam *config,
         return NULL;
     }
 
-    flb_info("[aws_msk_iam] build_msk_iam_payload_with_creds: generating payload for host: %s, region: %s",
-             host, config->region);
+    flb_debug("[aws_msk_iam] build_msk_iam_payload: generating payload for host: %s, region: %s",
+              host, config->region);
 
     /* Validate credentials */
     if (!creds) {
-        flb_error("[aws_msk_iam] build_msk_iam_payload_with_creds: credentials are NULL");
+        flb_error("[aws_msk_iam] build_msk_iam_payload: credentials are NULL");
         return NULL;
     }
 
     if (!creds->access_key_id || !creds->secret_access_key) {
-        flb_error("[aws_msk_iam] build_msk_iam_payload_with_creds: incomplete credentials");
+        flb_error("[aws_msk_iam] build_msk_iam_payload: incomplete credentials");
         return NULL;
     }
 
@@ -635,12 +635,19 @@ static void oauthbearer_token_refresh_cb(rd_kafka_t *rk,
         flb_debug("[aws_msk_iam] using MSK generic endpoint: %s", host);
     }
 
-    flb_info("[aws_msk_iam] requesting MSK IAM payload for region: %s, host: %s", config->region, host);
+    flb_debug("[aws_msk_iam] requesting MSK IAM payload for region: %s, host: %s", config->region, host);
 
     /* 
-     * Get credentials from provider. The provider handles caching and expiration internally.
-     * The provider automatically manages credential refresh when needed.
+     * Refresh credentials before generating OAuth token.
+     * This is necessary because provider's passive refresh only triggers when
+     * get_credentials is called and detects expiration. However, OAuth tokens
+     * are refreshed every ~15 minutes while IAM credentials expire after ~1 hour.
+     * If OAuth callbacks are spaced far apart, the passive refresh may not trigger
+     * before credentials expire, causing authentication failures.
      */
+    config->provider->provider_vtable->refresh(config->provider);
+
+    /* Get credentials from provider */
     creds = config->provider->provider_vtable->get_credentials(config->provider);
     if (!creds) {
         flb_error("[aws_msk_iam] failed to get AWS credentials from provider");