feat: require v2.3 of rdkafka for oauth bearer token refresh cb

conf.c

@@ -66,6 +66,8 @@ void kafka_conf_callbacks_dtor(kafka_conf_callbacks *cbs) /* {{{ */
     cbs->offset_commit = NULL;
     kafka_conf_callback_dtor(cbs->log);
     cbs->log = NULL;
+    kafka_conf_callback_dtor(cbs->oauthbearer_token_refresh);
+    cbs->oauthbearer_token_refresh = NULL;
 } /* }}} */
 
 static void kafka_conf_callback_copy(kafka_conf_callback **to, kafka_conf_callback *from) /* {{{ */
@@ -738,6 +740,11 @@ PHP_METHOD(RdKafka_Conf, setLogCb)
    Set token refresh callback for OAUTHBEARER sasl */
 PHP_METHOD(RdKafka_Conf, setOauthbearerTokenRefreshCb)
 {
+#ifndef HAS_RD_KAFKA_PARTITIONER_MURMUR2
+    zend_throw_exception_ex(NULL, 0, "This version of rdkafka does not support the OAUTHBEARER sasl mechanism");
+    return;
+#endif
+
     zend_fcall_info fci;
     zend_fcall_info_cache fcc;
     kafka_conf_object *conf;

config.m4

@@ -84,6 +84,12 @@ if test "$PHP_RDKAFKA" != "no"; then
     AC_MSG_WARN([murmur2 partitioner is not available])
   ])
 
+  AC_CHECK_LIB($LIBNAME,[rd_kafka_conf_set_oauthbearer_token_refresh_cb],[
+    AC_DEFINE(HAS_RD_KAFKA_OAUTHBEARER_TOKEN_REFRESH_CB,1,[ ])
+  ],[
+    AC_MSG_WARN([oauthbearer token refresh cb is not available])
+  ])
+
   LDFLAGS="$ORIG_LDFLAGS"
   CPPFLAGS="$ORIG_CPPFLAGS"
 

tests/conf_callback_oauthbearer.phpt

@@ -0,0 +1,18 @@
+--TEST--
+RdKafka\Conf
+--SKIPIF--
+<?php
+RD_KAFKA_VERSION >= 0x230000 || die("skip librdkafka too old");
+--FILE--
+<?php
+
+$conf = new RdKafka\Conf();
+
+echo "Setting oauth token bearer callback\n";
+$conf->setOauthbearerTokenRefreshCb(function () {});
+$dump = $conf->dump();
+var_dump(isset($dump["oauthbearer_token_refresh_cb"]));
+
+--EXPECT--
+Setting oauth token bearer callback
+bool(true)

tests/conf_callbacks.phpt

@@ -24,9 +24,11 @@ $dump = $conf->dump();
 var_dump(isset($dump["rebalance_cb"]));
 
 echo "Setting oauth token bearer callback\n";
-$conf->setOauthbearerTokenRefreshCb(function () {});
-$dump = $conf->dump();
-var_dump(isset($dump["oauthbearer_token_refresh_cb"]));
+try {
+    $conf->setOauthbearerTokenRefreshCb(function () {});
+} catch (Exception $e) {
+    echo $e->getMessage();
+}
 
 --EXPECT--
 Setting consume callback
@@ -36,4 +38,4 @@ bool(true)
 Setting rebalance callback
 bool(true)
 Setting oauth token bearer callback
-bool(true)
+This version of rdkafka does not support the OAUTHBEARER sasl mechanism