Skip to content

feat: finalize OAUTHBEARER support #547

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 27 commits into from
Oct 24, 2024
Merged

feat: finalize OAUTHBEARER support #547

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
ce6d3e9
feat: add support for oauthbearer_set_token/set_token_failure
cb-freddysart Jan 10, 2024
c0a98e7
fix: back out unrelated arginfo changes
cb-freddysart Jan 10, 2024
ee02108
fix: further arginfo fixes
cb-freddysart Jan 10, 2024
ad773d0
fix: handle strict errors
cb-freddysart Jan 10, 2024
12cb781
fix: remove comment
cb-freddysart Jan 15, 2024
410bccb
fix: mem leak around extensions
cb-freddysart Jan 18, 2024
8441345
fix: add length param for errstr
cb-freddysart Jan 18, 2024
363a9e9
fix: remove comment
cb-freddysart Jan 18, 2024
9d9e823
fix: swap to h, rename extension_size to extensions_size for consistency
cb-freddysart Jan 18, 2024
41c5d1f
fix: const cast
cb-freddysart Jan 18, 2024
4d3f493
Implement integration tests for oauth token methods
scorgn Oct 13, 2024
edead1c
Fix typo in kafka_ready
scorgn Oct 13, 2024
b92336b
Skip oauthbearer tests and setup steps if librdkafka version does not…
scorgn Oct 13, 2024
b86825b
Ensure tests compatible with all php versions
scorgn Oct 13, 2024
42d3110
Fix RD_KAFKA_VERSION comparison
scorgn Oct 14, 2024
9079b9a
Remove usage of json_encode as we don't have access to json extension
scorgn Oct 14, 2024
28cc1cc
Add newline to end of test file
scorgn Oct 14, 2024
2b859d6
Change test name to match contents
scorgn Oct 14, 2024
2a78a6c
Change test name to match contents
scorgn Oct 14, 2024
73acf80
Merge remote-tracking branch 'origin/6.x' into feat/oauthbearer-set-t…
scorgn Oct 14, 2024
32e3dfb
Change controller_id test to use 1 instead of 1001
scorgn Oct 14, 2024
be78aa0
Merge pull request #2 from scorgn/feat/oauthbearer-set-token-solve-co…
cb-freddysart Oct 15, 2024
82e320c
Merge branch 'feat/oauthbearer-set-token' of github.com:cb-freddysart…
cb-freddysart Oct 15, 2024
da58f00
feat: bump upload-artifact GH action
cb-freddysart Oct 15, 2024
976fc01
feat: add oauthbearer integration test to package.xml
cb-freddysart Oct 15, 2024
f497c3e
feat: bump metadata timeout
cb-freddysart Oct 16, 2024
3231fe2
feat: debug broker startup
cb-freddysart Oct 16, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/package.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See https://github.blog/changelog/2024-02-13-deprecation-notice-v1-and-v2-of-the-artifact-actions/

with:
path: 'php-rdkafka/rdkafka.tgz'
12 changes: 12 additions & 0 deletions .github/workflows/test.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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:
Expand Down
60 changes: 53 additions & 7 deletions .github/workflows/test/start-kafka.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
3 changes: 2 additions & 1 deletion conf.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion conf.stub.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
16 changes: 12 additions & 4 deletions conf_arginfo.h
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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()
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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);
Expand All @@ -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
Expand All @@ -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;
}
Expand All @@ -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;
}
16 changes: 12 additions & 4 deletions conf_legacy_arginfo.h
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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()
Expand Down Expand Up @@ -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()
Expand All @@ -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);
Expand All @@ -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
Expand All @@ -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;
}
Expand All @@ -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;
}
6 changes: 3 additions & 3 deletions config.m4
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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],[
Expand Down
1 change: 1 addition & 0 deletions package.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"/>
Expand Down
Loading
Loading