adding poll method to KafkaConsumer

Author: Rastusik

kafka_consumer.c

@@ -871,6 +871,26 @@ PHP_METHOD(RdKafka_KafkaConsumer, resumePartitions)
 }
 /* }}} */
 
+/* {{{ proto int RdKafka::poll(int $timeout_ms)
+   Polls the provided kafka handle for events */
+PHP_METHOD(RdKafka_KafkaConsumer, poll)
+{
+    object_intern *intern;
+    zend_long timeout;
+
+    if (zend_parse_parameters(ZEND_NUM_ARGS(), "l", &timeout) == FAILURE) {
+        return;
+    }
+
+    intern = get_object(getThis());
+    if (!intern) {
+        return;
+    }
+
+    RETURN_LONG(rd_kafka_poll(intern->rk, timeout));
+}
+/* }}} */
+
 #ifdef HAS_RD_KAFKA_OAUTHBEARER
 /* {{{ proto void RdKafka\KafkaConsumer::oauthbearerSetToken(string $token_value, int $lifetime_ms, string $principal_name, array $extensions = [])
  * Set SASL/OAUTHBEARER token and metadata

kafka_consumer.stub.php

@@ -82,6 +82,9 @@ public function pausePartitions(array $topic_partitions): array {}
     /** @tentative-return-type */
     public function resumePartitions(array $topic_partitions): array {}
 
+    /** @tentative-return-type */
+    public function poll(int $timeout_ms): int {}
+
     #ifdef HAS_RD_KAFKA_OAUTHBEARER
     /** @tentative-return-type */
     public function oauthbearerSetToken(string $token_value, int $lifetime_ms, string $principal_name, array $extensions = []): void {}

kafka_consumer_arginfo.h

@@ -129,6 +129,14 @@ ZEND_BEGIN_ARG_INFO_EX(arginfo_class_RdKafka_KafkaConsumer_queryWatermarkOffsets
 	ZEND_ARG_TYPE_INFO(0, timeout_ms, IS_LONG, 0)
 ZEND_END_ARG_INFO()
 
+#if (PHP_VERSION_ID >= 80100)
+ZEND_BEGIN_ARG_WITH_TENTATIVE_RETURN_TYPE_INFO_EX(arginfo_class_RdKafka_KafkaConsumer_poll, 0, 1, IS_LONG, 0)
+#else
+ZEND_BEGIN_ARG_INFO_EX(arginfo_class_RdKafka_KafkaConsumer_poll, 0, 0, 1)
+#endif
+	ZEND_ARG_TYPE_INFO(0, timeout_ms, IS_LONG, 0)
+ZEND_END_ARG_INFO()
+
 #if defined(HAS_RD_KAFKA_OAUTHBEARER)
 #if (PHP_VERSION_ID >= 80100)
 ZEND_BEGIN_ARG_WITH_TENTATIVE_RETURN_TYPE_INFO_EX(arginfo_class_RdKafka_KafkaConsumer_oauthbearerSetToken, 0, 3, IS_VOID, 0)
@@ -183,6 +191,7 @@ ZEND_METHOD(RdKafka_KafkaConsumer, queryWatermarkOffsets);
 ZEND_METHOD(RdKafka_KafkaConsumer, offsetsForTimes);
 ZEND_METHOD(RdKafka_KafkaConsumer, pausePartitions);
 ZEND_METHOD(RdKafka_KafkaConsumer, resumePartitions);
+ZEND_METHOD(RdKafka_KafkaConsumer, poll);
 #if defined(HAS_RD_KAFKA_OAUTHBEARER)
 ZEND_METHOD(RdKafka_KafkaConsumer, oauthbearerSetToken);
 ZEND_METHOD(RdKafka_KafkaConsumer, oauthbearerSetTokenFailure);
@@ -216,6 +225,7 @@ static const zend_function_entry class_RdKafka_KafkaConsumer_methods[] = {
 	ZEND_ME(RdKafka_KafkaConsumer, offsetsForTimes, arginfo_class_RdKafka_KafkaConsumer_offsetsForTimes, ZEND_ACC_PUBLIC)
 	ZEND_ME(RdKafka_KafkaConsumer, pausePartitions, arginfo_class_RdKafka_KafkaConsumer_pausePartitions, ZEND_ACC_PUBLIC)
 	ZEND_ME(RdKafka_KafkaConsumer, resumePartitions, arginfo_class_RdKafka_KafkaConsumer_resumePartitions, ZEND_ACC_PUBLIC)
+	ZEND_ME(RdKafka_KafkaConsumer, poll, arginfo_class_RdKafka_KafkaConsumer_poll, ZEND_ACC_PUBLIC)
 #if defined(HAS_RD_KAFKA_OAUTHBEARER)
 	ZEND_ME(RdKafka_KafkaConsumer, oauthbearerSetToken, arginfo_class_RdKafka_KafkaConsumer_oauthbearerSetToken, ZEND_ACC_PUBLIC)
 	ZEND_ME(RdKafka_KafkaConsumer, oauthbearerSetTokenFailure, arginfo_class_RdKafka_KafkaConsumer_oauthbearerSetTokenFailure, ZEND_ACC_PUBLIC)

tests/oauthbearer_integration_hl.phpt

@@ -60,14 +60,6 @@ $producer->poll(0);
 $topicName = sprintf("test_rdkafka_%s", uniqid());
 $topic = $producer->newTopic($topicName);
 
-try {
-    $producer->getMetadata(false, $topic, 10*1000);
-    echo "Metadata retrieved successfully when refresh callback set token\n";
-} catch (\RdKafka\Exception $e) {
-    echo "FAIL: Caught exception when getting metadata after successfully refreshing any token:\n";
-    printf("%s: %s\n", get_class($e), $e->getMessage());
-}
-
 echo "Writing test data\n";
 $topic->produce(RD_KAFKA_PARTITION_UA, 0, "Test");
 $producer->poll(0);
@@ -89,7 +81,7 @@ $confConsumer->setErrorCb(function ($producer, $err, $errstr) {
     printf("%s: %s\n", rd_kafka_err2str($err), $errstr);
 });
 
-// Test that refresh token with setting token accurately will succeed when getting metadata
+// Test that refresh token with setting token accurately will succeed when consuming data
 $confConsumer->setOauthbearerTokenRefreshCb(function ($consumer) {
     echo "Refreshing token and succeeding\n";
     $token = generateJws();
@@ -118,9 +110,26 @@ $message = $consumer->consume(500);
 echo $message->err === -185 ? "Received empty message when reading data after not setting or refreshing any token\n" :
     "FAIL: Did receive a message after not setting or refreshing any token\n";
 
+// Test that metadata will be loaded before data consumption, under the condition that poll is called
+$confConsumer->setOauthbearerTokenRefreshCb(function ($consumer) {
+    echo "Refreshing token on poll and succeeding\n";
+    $token = generateJws();
+    $consumer->oauthbearerSetToken($token['value'], (string) $token['expiryMs'], $token['principal']);
+});
+$consumer = new \RdKafka\KafkaConsumer($confConsumer);
+$consumerTopic = $consumer->newTopic($topicName);
+$consumer->poll(0);
+
+try {
+    echo "Reading metadata\n";
+    $consumer->getMetadata(false, $consumerTopic, 1000);
+    echo "Metadata was fetched successfully after calling poll\n";
+} catch (\RdKafka\Exception $e) {
+    echo "FAIL: Caught exception when getting metadata after calling poll\n";
+}
+
 --EXPECT--
 Refreshing token and succeeding
-Metadata retrieved successfully when refresh callback set token
 Writing test data
 Write successful
 Reading data
@@ -131,3 +140,6 @@ Reading data
 Setting token failure in refresh cb
 Local: Authentication failure: Failed to acquire SASL OAUTHBEARER token: Token failure before data consumption
 Received empty message when reading data after not setting or refreshing any token
+Refreshing token on poll and succeeding
+Reading metadata
+Metadata was fetched successfully after calling poll