From 122678e881de94721458fd948f38e65366b68689 Mon Sep 17 00:00:00 2001
From: Tyler Rockwood <rockwood@openai.com>
Date: Sat, 8 Jun 2024 03:32:13 +0000
Subject: [PATCH] Fix some OAuth related issues

* Add logging for config issues
* oauth callback expiration should be milliseconds not microseconds
* Return on error early during oauth callback
* Prevent no config from throwing exception constructing string
---
 include/kafka/ClientCommon.h | 3 ++-
 include/kafka/KafkaClient.h  | 5 +++--
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/include/kafka/ClientCommon.h b/include/kafka/ClientCommon.h
index c399d4d0..c0ff7e8f 100644
--- a/include/kafka/ClientCommon.h
+++ b/include/kafka/ClientCommon.h
@@ -1,5 +1,6 @@
 #pragma once
 
+#include <chrono>
 #include <kafka/Project.h>
 
 #include <kafka/Error.h>
@@ -32,7 +33,7 @@ namespace KAFKA_API { namespace clients {
         using KeyValuePairs = std::map<std::string, std::string>;
 
         std::string               value;
-        std::chrono::microseconds mdLifetime{};
+        std::chrono::milliseconds mdLifetime{};
         std::string               mdPrincipalName;
         KeyValuePairs             extensions;
     };
diff --git a/include/kafka/KafkaClient.h b/include/kafka/KafkaClient.h
index 36d9a368..6d07d824 100644
--- a/include/kafka/KafkaClient.h
+++ b/include/kafka/KafkaClient.h
@@ -372,7 +372,7 @@ KafkaClient::KafkaClient(ClientType                     clientType,
         }
         else
         {
-            KAFKA_API_DO_LOG(Log::Level::Err, "failed to be initialized with property[%s:%s], result[%d]", k.c_str(), v->c_str(), result);
+            KAFKA_API_DO_LOG(Log::Level::Err, "failed to be initialized with property[%s:%s], result[%d]: %s", k.c_str(), v->c_str(), result, errInfo.c_str());
         }
     }
 
@@ -570,11 +570,12 @@ KafkaClient::oauthbearerTokenRefreshCallback(rd_kafka_t* rk, const char* oauthbe
 
     try
     {
-        oauthbearerToken = kafkaClient(rk).onOauthbearerTokenRefresh(oauthbearerConfig);
+        oauthbearerToken = kafkaClient(rk).onOauthbearerTokenRefresh(oauthbearerConfig != nullptr ? oauthbearerConfig : "");
     }
     catch (const std::exception& e)
     {
         rd_kafka_oauthbearer_set_token_failure(rk, e.what());
+        return;
     }
 
     LogBuffer<LOG_BUFFER_SIZE> errInfo;