feat: finalize OAUTHBEARER support (#547)

Co-authored-by: Sarina Corrigan <26685104+scorgn@users.noreply.github.com>

Author: cb-freddysart

.github/workflows/package.yml

@@ -20,6 +20,6 @@ jobs:
               run: './php-rdkafka/.github/workflows/package/package.sh'
 
             - name: 'Archive package'
-              uses: 'actions/upload-artifact@v2'
+              uses: 'actions/upload-artifact@v4'
               with:
                   path: 'php-rdkafka/rdkafka.tgz'

.github/workflows/test.yml

@@ -109,28 +109,39 @@ jobs:
               # librdkafka 1.0.1
               - php: '8.1.0'
                 librdkafka: 'v1.0.1'
+                skipoauth: '1'
               - php: '8.0.0'
                 librdkafka: 'v1.0.1'
+                skipoauth: '1'
               - php: '7.4.0'
                 librdkafka: 'v1.0.1'
+                skipoauth: '1'
               - php: '7.3.0'
                 librdkafka: 'v1.0.1'
+                skipoauth: '1'
 
               # librdkafka 0.11.6
               - php: '8.1.0'
                 librdkafka: 'v0.11.6'
+                skipoauth: '1'
               - php: '8.0.0'
                 librdkafka: 'v0.11.6'
+                skipoauth: '1'
               - php: '7.4.0'
                 librdkafka: 'v0.11.6'
+                skipoauth: '1'
               - php: '7.3.0'
                 librdkafka: 'v0.11.6'
+                skipoauth: '1'
               - php: '7.2.0'
                 librdkafka: 'v0.11.6'
+                skipoauth: '1'
               - php: '7.1.0'
                 librdkafka: 'v0.11.6'
+                skipoauth: '1'
               - php: '7.0.0'
                 librdkafka: 'v0.11.6'
+                skipoauth: '1'
 
               # librdkafka master (experimental, does not block PRs)
               - php: '8.3.0'
@@ -158,6 +169,7 @@ jobs:
             PHP_VERSION: ${{ matrix.php }}
             LIBRDKAFKA_VERSION: ${{ matrix.librdkafka }}
             MEMORY_CHECK: ${{ matrix.memcheck }}
+            SKIP_OAUTH: ${{ matrix.skipoauth }}
             TEST_KAFKA_BROKERS: kafka:9092
             TEST_KAFKA_BROKER_VERSION: 2.6
         steps:

.github/workflows/test/start-kafka.sh

@@ -4,17 +4,63 @@ docker network create kafka_network
 docker pull wurstmeister/zookeeper:latest
 docker run -d --network kafka_network --name zookeeper wurstmeister/zookeeper:latest
 docker pull wurstmeister/kafka:latest
-docker run -d -p 9092:9092 --network kafka_network -e "KAFKA_AUTO_CREATE_TOPICS_ENABLE=true" -e "KAFKA_CREATE_TOPICS=test-topic:1:1:compact" -e "KAFKA_ADVERTISED_HOST_NAME=kafka" -e "KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181" -e "KAFKA_ADVERTISED_PORT=9092" --name kafka wurstmeister/kafka:latest
-printf "\n127.0.0.1  kafka\n"|sudo tee /etc/hosts >/dev/null
+docker run -d -p 9092:9092 --network kafka_network \
+    -e "KAFKA_AUTO_CREATE_TOPICS_ENABLE=true" \
+    -e "KAFKA_CREATE_TOPICS=test-topic:1:1:compact" \
+    -e "KAFKA_BROKER_ID=1" \
+    -e "KAFKA_ADVERTISED_HOST_NAME=kafka" \
+    -e "KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181/kafka" \
+    -e "KAFKA_ADVERTISED_PORT=9092" \
+    --name kafka wurstmeister/kafka:latest
 
-echo "Waiting for Kafka to be ready"
+if [ ${SKIP_OAUTH:-0} -ne 1 ]; then
+  docker run -d -p 29092:29092 --network kafka_network \
+      -e "KAFKA_AUTO_CREATE_TOPICS_ENABLE=true" \
+      -e "KAFKA_CREATE_TOPICS=test-topic:1:1:compact" \
+      -e "KAFKA_ADVERTISED_HOST_NAME=kafka_oauth2" \
+      -e "KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181/kafka_oauth2" \
+      -e "KAFKA_ADVERTISED_PORT=29092" \
+      -e "KAFKA_BROKER_ID=2" \
+      -e "KAFKA_LISTENERS=SASLPLAINTEXT://kafka_oauth2:29092" \
+      -e "KAFKA_ADVERTISED_LISTENERS=SASLPLAINTEXT://kafka_oauth2:29092" \
+      -e "KAFKA_LISTENER_NAME_SASLPLAINTEXT_OAUTHBEARER_SASL_JAAS_CONFIG=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required unsecuredValidatorRequiredScope=\"required-scope\" unsecuredLoginStringClaim_scope=\"required-scope\" unsecuredLoginStringClaim_sub=\"admin\";" \
+      -e "KAFKA_INTER_BROKER_LISTENER_NAME=SASLPLAINTEXT" \
+      -e "KAFKA_SASL_ENABLED_MECHANISMS=OAUTHBEARER" \
+      -e "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=SASLPLAINTEXT:SASL_PLAINTEXT" \
+      -e "KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL=OAUTHBEARER" \
+      --name kafka_oauth2 wurstmeister/kafka:latest
+fi
+
+printf "\n127.0.0.1  kafka\n127.0.0.1  kafka_oauth2\n"|sudo tee /etc/hosts >/dev/null
+
+echo "Waiting for Kafka services to be ready"
+
+kafka_ready=0
+kafka_oauth2_ready=0
 
 for i in $(seq 1 20); do
-    if kafkacat -b 127.0.0.1 -L; then
-        echo "Kafka is ready"
-        exit 0
+    if [ $kafka_ready -eq 0 ]; then
+        if kafkacat -b kafka:9092 -L -m 30 -d broker; then
+            kafka_ready=1
+            echo "Kafka is ready"
+        fi
+    fi
+    if [ $kafka_oauth2_ready -eq 0 ] && [ ${SKIP_OAUTH:-0} -ne 1 ]; then
+        if kafkacat -b kafka_oauth2:29092 \
+            -X security.protocol=SASL_PLAINTEXT \
+            -X sasl.mechanisms=OAUTHBEARER \
+            -X enable.sasl.oauthbearer.unsecure.jwt="true" \
+            -X sasl.oauthbearer.config="principal=admin scope=required-scope" -L
+        then
+            kafka_oauth2_ready=1
+            echo "Kafka OAuth2 is ready"
+        fi
+    fi
+
+    if [ $kafka_ready -eq 1 ] && ( [ $kafka_oauth2_ready -eq 1 ] || [ ${SKIP_OAUTH:-0} -eq 1 ] ); then
+      exit 0
     fi
 done
 
-echo "Timedout waiting for Kafka to be ready"
+echo "Timedout waiting for Kafka services to be ready"
 exit 1

conf.c

@@ -81,6 +81,7 @@ static void kafka_conf_callback_copy(kafka_conf_callback **to, kafka_conf_callba
 
 void kafka_conf_callbacks_copy(kafka_conf_callbacks *to, kafka_conf_callbacks *from) /* {{{ */
 {
+    kafka_conf_callback_copy(&to->oauthbearer_token_refresh, from->oauthbearer_token_refresh);
     kafka_conf_callback_copy(&to->error, from->error);
     kafka_conf_callback_copy(&to->rebalance, from->rebalance);
     kafka_conf_callback_copy(&to->dr_msg, from->dr_msg);
@@ -734,7 +735,7 @@ PHP_METHOD(RdKafka_Conf, setLogCb)
 }
 /* }}} */
 
-#ifdef HAS_RD_KAFKA_OAUTHBEARER_TOKEN_REFRESH_CB
+#ifdef HAS_RD_KAFKA_OAUTHBEARER
 /* {{{ proto void RdKafka\Conf::setOauthbearerTokenRefreshCb(mixed $callback)
    Set token refresh callback for OAUTHBEARER sasl */
 PHP_METHOD(RdKafka_Conf, setOauthbearerTokenRefreshCb)

conf.stub.php

@@ -45,7 +45,7 @@ public function setOffsetCommitCb(callable $callback): void {}
     /** @tentative-return-type */
     public function setLogCb(callable $callback): void {}
 
-    #ifdef HAS_RD_KAFKA_OAUTHBEARER_TOKEN_REFRESH_CB
+    #ifdef HAS_RD_KAFKA_OAUTHBEARER
     /** @tentative-return-type */
     public function setOauthbearerTokenRefreshCb(callable $callback): void {}
     #endif

conf_arginfo.h

@@ -1,5 +1,5 @@
 /* This is a generated file, edit the .stub.php file instead.
- * Stub hash: 86934f54199c0af37cadfcedafeaffb569e33beb */
+ * Stub hash: a72d2e1796ed7f89185f543973c659a6a704f347 */
 
 ZEND_BEGIN_ARG_INFO_EX(arginfo_class_RdKafka_Conf___construct, 0, 0, 0)
 ZEND_END_ARG_INFO()
@@ -48,7 +48,7 @@ ZEND_END_ARG_INFO()
 
 #define arginfo_class_RdKafka_Conf_setLogCb arginfo_class_RdKafka_Conf_setErrorCb
 
-#if defined(HAS_RD_KAFKA_OAUTHBEARER_TOKEN_REFRESH_CB)
+#if defined(HAS_RD_KAFKA_OAUTHBEARER)
 #if (PHP_VERSION_ID >= 80100)
 ZEND_BEGIN_ARG_WITH_TENTATIVE_RETURN_TYPE_INFO_EX(arginfo_class_RdKafka_Conf_setOauthbearerTokenRefreshCb, 0, 1, IS_VOID, 0)
 #else
@@ -83,7 +83,7 @@ ZEND_METHOD(RdKafka_Conf, setRebalanceCb);
 ZEND_METHOD(RdKafka_Conf, setConsumeCb);
 ZEND_METHOD(RdKafka_Conf, setOffsetCommitCb);
 ZEND_METHOD(RdKafka_Conf, setLogCb);
-#if defined(HAS_RD_KAFKA_OAUTHBEARER_TOKEN_REFRESH_CB)
+#if defined(HAS_RD_KAFKA_OAUTHBEARER)
 ZEND_METHOD(RdKafka_Conf, setOauthbearerTokenRefreshCb);
 #endif
 ZEND_METHOD(RdKafka_TopicConf, __construct);
@@ -101,7 +101,7 @@ static const zend_function_entry class_RdKafka_Conf_methods[] = {
 	ZEND_ME(RdKafka_Conf, setConsumeCb, arginfo_class_RdKafka_Conf_setConsumeCb, ZEND_ACC_PUBLIC)
 	ZEND_ME(RdKafka_Conf, setOffsetCommitCb, arginfo_class_RdKafka_Conf_setOffsetCommitCb, ZEND_ACC_PUBLIC)
 	ZEND_ME(RdKafka_Conf, setLogCb, arginfo_class_RdKafka_Conf_setLogCb, ZEND_ACC_PUBLIC)
-#if defined(HAS_RD_KAFKA_OAUTHBEARER_TOKEN_REFRESH_CB)
+#if defined(HAS_RD_KAFKA_OAUTHBEARER)
 	ZEND_ME(RdKafka_Conf, setOauthbearerTokenRefreshCb, arginfo_class_RdKafka_Conf_setOauthbearerTokenRefreshCb, ZEND_ACC_PUBLIC)
 #endif
 	ZEND_FE_END
@@ -128,7 +128,11 @@ static zend_class_entry *register_class_RdKafka_Conf(void)
 	zend_class_entry ce, *class_entry;
 
 	INIT_NS_CLASS_ENTRY(ce, "RdKafka", "Conf", class_RdKafka_Conf_methods);
+#if (PHP_VERSION_ID >= 80400)
+	class_entry = zend_register_internal_class_with_flags(&ce, NULL, 0);
+#else
 	class_entry = zend_register_internal_class_ex(&ce, NULL);
+#endif
 
 	return class_entry;
 }
@@ -138,7 +142,11 @@ static zend_class_entry *register_class_RdKafka_TopicConf(void)
 	zend_class_entry ce, *class_entry;
 
 	INIT_NS_CLASS_ENTRY(ce, "RdKafka", "TopicConf", class_RdKafka_TopicConf_methods);
+#if (PHP_VERSION_ID >= 80400)
+	class_entry = zend_register_internal_class_with_flags(&ce, NULL, 0);
+#else
 	class_entry = zend_register_internal_class_ex(&ce, NULL);
+#endif
 
 	return class_entry;
 }

conf_legacy_arginfo.h

@@ -1,5 +1,5 @@
 /* This is a generated file, edit the .stub.php file instead.
- * Stub hash: 86934f54199c0af37cadfcedafeaffb569e33beb */
+ * Stub hash: a72d2e1796ed7f89185f543973c659a6a704f347 */
 
 ZEND_BEGIN_ARG_INFO_EX(arginfo_class_RdKafka_Conf___construct, 0, 0, 0)
 ZEND_END_ARG_INFO()
@@ -32,7 +32,7 @@ ZEND_END_ARG_INFO()
 
 #define arginfo_class_RdKafka_Conf_setLogCb arginfo_class_RdKafka_Conf_setErrorCb
 
-#if defined(HAS_RD_KAFKA_OAUTHBEARER_TOKEN_REFRESH_CB)
+#if defined(HAS_RD_KAFKA_OAUTHBEARER)
 ZEND_BEGIN_ARG_INFO_EX(arginfo_class_RdKafka_Conf_setOauthbearerTokenRefreshCb, 0, 0, 1)
 	ZEND_ARG_INFO(0, callback)
 ZEND_END_ARG_INFO()
@@ -59,7 +59,7 @@ ZEND_METHOD(RdKafka_Conf, setRebalanceCb);
 ZEND_METHOD(RdKafka_Conf, setConsumeCb);
 ZEND_METHOD(RdKafka_Conf, setOffsetCommitCb);
 ZEND_METHOD(RdKafka_Conf, setLogCb);
-#if defined(HAS_RD_KAFKA_OAUTHBEARER_TOKEN_REFRESH_CB)
+#if defined(HAS_RD_KAFKA_OAUTHBEARER)
 ZEND_METHOD(RdKafka_Conf, setOauthbearerTokenRefreshCb);
 #endif
 ZEND_METHOD(RdKafka_TopicConf, __construct);
@@ -77,7 +77,7 @@ static const zend_function_entry class_RdKafka_Conf_methods[] = {
 	ZEND_ME(RdKafka_Conf, setConsumeCb, arginfo_class_RdKafka_Conf_setConsumeCb, ZEND_ACC_PUBLIC)
 	ZEND_ME(RdKafka_Conf, setOffsetCommitCb, arginfo_class_RdKafka_Conf_setOffsetCommitCb, ZEND_ACC_PUBLIC)
 	ZEND_ME(RdKafka_Conf, setLogCb, arginfo_class_RdKafka_Conf_setLogCb, ZEND_ACC_PUBLIC)
-#if defined(HAS_RD_KAFKA_OAUTHBEARER_TOKEN_REFRESH_CB)
+#if defined(HAS_RD_KAFKA_OAUTHBEARER)
 	ZEND_ME(RdKafka_Conf, setOauthbearerTokenRefreshCb, arginfo_class_RdKafka_Conf_setOauthbearerTokenRefreshCb, ZEND_ACC_PUBLIC)
 #endif
 	ZEND_FE_END
@@ -104,7 +104,11 @@ static zend_class_entry *register_class_RdKafka_Conf(void)
 	zend_class_entry ce, *class_entry;
 
 	INIT_NS_CLASS_ENTRY(ce, "RdKafka", "Conf", class_RdKafka_Conf_methods);
+#if (PHP_VERSION_ID >= 80400)
+	class_entry = zend_register_internal_class_with_flags(&ce, NULL, 0);
+#else
 	class_entry = zend_register_internal_class_ex(&ce, NULL);
+#endif
 
 	return class_entry;
 }
@@ -114,7 +118,11 @@ static zend_class_entry *register_class_RdKafka_TopicConf(void)
 	zend_class_entry ce, *class_entry;
 
 	INIT_NS_CLASS_ENTRY(ce, "RdKafka", "TopicConf", class_RdKafka_TopicConf_methods);
+#if (PHP_VERSION_ID >= 80400)
+	class_entry = zend_register_internal_class_with_flags(&ce, NULL, 0);
+#else
 	class_entry = zend_register_internal_class_ex(&ce, NULL);
+#endif
 
 	return class_entry;
 }

config.m4

@@ -94,10 +94,10 @@ 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_CHECK_LIB($LIBNAME,[rd_kafka_oauthbearer_set_token],[
+    AC_DEFINE(HAS_RD_KAFKA_OAUTHBEARER,1,[ ])
   ],[
-    AC_MSG_WARN([oauthbearer token refresh cb is not available])
+    AC_MSG_WARN([oauthbearer support is not available])
   ])
 
   AC_CHECK_LIB($LIBNAME,[rd_kafka_incremental_assign, rd_kafka_incremental_unassign],[

package.xml

@@ -132,6 +132,7 @@
     <file role="test" name="metadata_partition_001.phpt"/>
     <file role="test" name="metadata_topic_001.phpt"/>
     <file role="test" name="new_topic_with_conf.phpt"/>
+    <file role="test" name="oauthbearer_integration.phpt"/>
     <file role="test" name="pause_resume.phpt"/>
     <file role="test" name="produce_consume.phpt"/>
     <file role="test" name="produce_consume_queue.phpt"/>