You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "shibd (via GitHub)" <gi...@apache.org> on 2023/12/05 01:24:03 UTC
[PR] feat: Support batch receive for consumer. [pulsar-client-node]
shibd opened a new pull request, #357:
URL: https://github.com/apache/pulsar-client-node/pull/357
Master Issue: #274
### Motivation
#274 Support batch receive for the consumer.
### Modifications
- Add batch receive policy config on the consumer.
- Add batch receive interface on the consumer.
- Refactor consumer `config set` and `verify logic`.
- Since config set logic with verify logic for the consumer is separate, it will be more complicated to read the code and subsequent extensions.
### Verifying this change
- Add batch receive unit test on Consumer.
- Add more config verify unit test on Consumer.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [x] `doc-required`
(Your PR needs to update docs and you will update later)
- [ ] `doc-not-needed`
(Please explain why)
- [ ] `doc`
(Your PR contains doc changes)
- [ ] `doc-complete`
(Docs have been already added)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1441584431
##########
src/Consumer.cc:
##########
@@ -192,36 +193,17 @@ struct ConsumerNewInstanceContext {
Napi::Value Consumer::NewInstance(const Napi::CallbackInfo &info, std::shared_ptr<pulsar_client_t> cClient) {
auto deferred = ThreadSafeDeferred::New(info.Env());
auto config = info[0].As<Napi::Object>();
- std::shared_ptr<ConsumerConfig> consumerConfig = std::make_shared<ConsumerConfig>(config, &MessageListener);
+ std::shared_ptr<ConsumerConfig> consumerConfig = std::make_shared<ConsumerConfig>();
+
+ consumerConfig->InitConfig(deferred, config, &MessageListener);
+ if (deferred->IsDone()) {
+ return deferred->Promise();
+ }
const std::string &topic = consumerConfig->GetTopic();
const std::vector<std::string> &topics = consumerConfig->GetTopics();
const std::string &topicsPattern = consumerConfig->GetTopicsPattern();
- if (topic.empty() && topics.size() == 0 && topicsPattern.empty()) {
- deferred->Reject(
- std::string("Topic, topics or topicsPattern is required and must be specified as a string when "
- "creating consumer"));
- return deferred->Promise();
- }
const std::string &subscription = consumerConfig->GetSubscription();
- if (subscription.empty()) {
- deferred->Reject(
- std::string("Subscription is required and must be specified as a string when creating consumer"));
- return deferred->Promise();
- }
- int32_t ackTimeoutMs = consumerConfig->GetAckTimeoutMs();
- if (ackTimeoutMs != 0 && ackTimeoutMs < MIN_ACK_TIMEOUT_MILLIS) {
- std::string msg("Ack timeout should be 0 or greater than or equal to " +
- std::to_string(MIN_ACK_TIMEOUT_MILLIS));
- deferred->Reject(msg);
- return deferred->Promise();
- }
- int32_t nAckRedeliverTimeoutMs = consumerConfig->GetNAckRedeliverTimeoutMs();
- if (nAckRedeliverTimeoutMs < 0) {
- std::string msg("NAck timeout should be greater than or equal to zero");
- deferred->Reject(msg);
- return deferred->Promise();
Review Comment:
I see. But what's the motivation to move it to `ConsumerConfig::InitConfig`? Is it related to this feature?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1444120780
##########
src/ConsumerConfig.cc:
##########
@@ -265,6 +291,39 @@ ConsumerConfig::ConsumerConfig(const Napi::Object &consumerConfig, pulsar_messag
}
pulsar_consumer_configuration_set_dlq_policy(this->cConsumerConfig.get(), &dlq_policy);
}
+
+ if (consumerConfig.Has(CFG_BATCH_RECEIVE_POLICY) &&
+ consumerConfig.Get(CFG_BATCH_RECEIVE_POLICY).IsObject()) {
+ Napi::Object propObj = consumerConfig.Get(CFG_BATCH_RECEIVE_POLICY).ToObject();
+ int maxNumMessages = -1;
+ if (propObj.Has(CFG_BATCH_RECEIVE_POLICY_MAX_NUM_MESSAGES) &&
Review Comment:
This logic has been added on the CPP client: https://github.com/apache/pulsar-client-cpp/blob/main/lib/ConsumerImplBase.cc#L38-L49
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1441469443
##########
src/Consumer.cc:
##########
@@ -192,36 +193,17 @@ struct ConsumerNewInstanceContext {
Napi::Value Consumer::NewInstance(const Napi::CallbackInfo &info, std::shared_ptr<pulsar_client_t> cClient) {
auto deferred = ThreadSafeDeferred::New(info.Env());
auto config = info[0].As<Napi::Object>();
- std::shared_ptr<ConsumerConfig> consumerConfig = std::make_shared<ConsumerConfig>(config, &MessageListener);
+ std::shared_ptr<ConsumerConfig> consumerConfig = std::make_shared<ConsumerConfig>();
+
+ consumerConfig->InitConfig(deferred, config, &MessageListener);
+ if (deferred->IsDone()) {
+ return deferred->Promise();
+ }
const std::string &topic = consumerConfig->GetTopic();
const std::vector<std::string> &topics = consumerConfig->GetTopics();
const std::string &topicsPattern = consumerConfig->GetTopicsPattern();
- if (topic.empty() && topics.size() == 0 && topicsPattern.empty()) {
- deferred->Reject(
- std::string("Topic, topics or topicsPattern is required and must be specified as a string when "
- "creating consumer"));
- return deferred->Promise();
- }
const std::string &subscription = consumerConfig->GetSubscription();
- if (subscription.empty()) {
- deferred->Reject(
- std::string("Subscription is required and must be specified as a string when creating consumer"));
- return deferred->Promise();
- }
- int32_t ackTimeoutMs = consumerConfig->GetAckTimeoutMs();
- if (ackTimeoutMs != 0 && ackTimeoutMs < MIN_ACK_TIMEOUT_MILLIS) {
- std::string msg("Ack timeout should be 0 or greater than or equal to " +
- std::to_string(MIN_ACK_TIMEOUT_MILLIS));
- deferred->Reject(msg);
- return deferred->Promise();
- }
- int32_t nAckRedeliverTimeoutMs = consumerConfig->GetNAckRedeliverTimeoutMs();
- if (nAckRedeliverTimeoutMs < 0) {
- std::string msg("NAck timeout should be greater than or equal to zero");
- deferred->Reject(msg);
- return deferred->Promise();
Review Comment:
Move to `consumerConfig->InitConfig` method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1444130734
##########
src/ConsumerConfig.cc:
##########
@@ -265,6 +291,39 @@ ConsumerConfig::ConsumerConfig(const Napi::Object &consumerConfig, pulsar_messag
}
pulsar_consumer_configuration_set_dlq_policy(this->cConsumerConfig.get(), &dlq_policy);
}
+
+ if (consumerConfig.Has(CFG_BATCH_RECEIVE_POLICY) &&
+ consumerConfig.Get(CFG_BATCH_RECEIVE_POLICY).IsObject()) {
+ Napi::Object propObj = consumerConfig.Get(CFG_BATCH_RECEIVE_POLICY).ToObject();
+ int maxNumMessages = -1;
+ if (propObj.Has(CFG_BATCH_RECEIVE_POLICY_MAX_NUM_MESSAGES) &&
Review Comment:
Got it, thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1441615671
##########
src/ConsumerConfig.h:
##########
@@ -21,14 +21,17 @@
#define CONSUMER_CONFIG_H
#include <pulsar/c/consumer_configuration.h>
+#include "ThreadSafeDeferred.h"
#include "MessageListener.h"
#define MIN_ACK_TIMEOUT_MILLIS 10000
class ConsumerConfig {
public:
- ConsumerConfig(const Napi::Object &consumerConfig, pulsar_message_listener messageListener);
+ ConsumerConfig();
~ConsumerConfig();
+ void InitConfig(const std::shared_ptr<ThreadSafeDeferred> deferred, const Napi::Object &consumerConfig,
Review Comment:
Don't pass const value as the parameter type. For example,
```c++
struct Foo {
void f1() const {}
void f2() {}
};
void f(const std::shared_ptr<Foo> foo) {
foo->f1();
foo->f2(); // OK, even though f2() is a non-const method
}
```
`f()` can only guarantee `foo` itself cannot be modified like `foo.set(nullptr)`.
However, since this method only needs a reference to the `ThreadSafeDeferred` object and does not transfer or share the ownership of `deferred` to another thread, I suggest passing a reference here.
```c++
void InitConfig(ThreadSafeDeferred &deferred, const Napi::Object &consumerConfig,
pulsar_message_listener messageListener);
```
Then call `deferred.Reject` instead of `deferred->Reject` in the implementation.
BTW, I still don't get why this refactoring is needed in this PR.
##########
tests/consumer.test.js:
##########
@@ -315,6 +346,47 @@ const Pulsar = require('../index');
consumer.close();
dlqConsumer.close();
});
+
+ test('Batch Receive', async () => {
+ const topicName = 'batch-receive-test-topic';
+ const producer = await client.createProducer({
+ topic: topicName,
+ });
+
+ const consumer = await client.subscribe({
+ topic: topicName,
+ subscription: 'sub1',
+ subscriptionType: 'Shared',
+ batchReceivePolicy: {
+ maxNumMessages: 10,
+ maxNumBytes: -1,
+ timeoutMs: 500,
Review Comment:
+1. You'd better test the `timeoutMs` config works.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1444114662
##########
src/ConsumerConfig.h:
##########
@@ -21,14 +21,17 @@
#define CONSUMER_CONFIG_H
#include <pulsar/c/consumer_configuration.h>
+#include "ThreadSafeDeferred.h"
#include "MessageListener.h"
#define MIN_ACK_TIMEOUT_MILLIS 10000
class ConsumerConfig {
public:
- ConsumerConfig(const Napi::Object &consumerConfig, pulsar_message_listener messageListener);
+ ConsumerConfig();
~ConsumerConfig();
+ void InitConfig(const std::shared_ptr<ThreadSafeDeferred> deferred, const Napi::Object &consumerConfig,
Review Comment:
Got it. Thanks for your reminder.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1440408093
##########
tests/consumer.test.js:
##########
@@ -315,6 +346,47 @@ const Pulsar = require('../index');
consumer.close();
dlqConsumer.close();
});
+
+ test('Batch Receive', async () => {
+ const topicName = 'batch-receive-test-topic';
+ const producer = await client.createProducer({
+ topic: topicName,
+ });
+
+ const consumer = await client.subscribe({
+ topic: topicName,
+ subscription: 'sub1',
+ subscriptionType: 'Shared',
+ batchReceivePolicy: {
+ maxNumMessages: 10,
+ maxNumBytes: -1,
+ timeoutMs: 500,
Review Comment:
Could we add a case? Such as sending messages less than 10 to verify timeout works well.
##########
src/ConsumerConfig.cc:
##########
@@ -265,6 +291,39 @@ ConsumerConfig::ConsumerConfig(const Napi::Object &consumerConfig, pulsar_messag
}
pulsar_consumer_configuration_set_dlq_policy(this->cConsumerConfig.get(), &dlq_policy);
}
+
+ if (consumerConfig.Has(CFG_BATCH_RECEIVE_POLICY) &&
+ consumerConfig.Get(CFG_BATCH_RECEIVE_POLICY).IsObject()) {
+ Napi::Object propObj = consumerConfig.Get(CFG_BATCH_RECEIVE_POLICY).ToObject();
+ int maxNumMessages = -1;
+ if (propObj.Has(CFG_BATCH_RECEIVE_POLICY_MAX_NUM_MESSAGES) &&
Review Comment:
I found Java client will use the `receiverQueueSize` if `maxNumMessages` is bigger than the `receiverQueueSize`, I'm not sure whether we can add this check in the node client.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1444121215
##########
tests/consumer.test.js:
##########
@@ -315,6 +346,47 @@ const Pulsar = require('../index');
consumer.close();
dlqConsumer.close();
});
+
+ test('Batch Receive', async () => {
+ const topicName = 'batch-receive-test-topic';
+ const producer = await client.createProducer({
+ topic: topicName,
+ });
+
+ const consumer = await client.subscribe({
+ topic: topicName,
+ subscription: 'sub1',
+ subscriptionType: 'Shared',
+ batchReceivePolicy: {
+ maxNumMessages: 10,
+ maxNumBytes: -1,
+ timeoutMs: 500,
Review Comment:
I will add it.
BTW: we shouldn't need to do tests with very complete boundaries on the node client, because it depends on the CPP client, and we've added tests for it in cpp
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower merged PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1444113403
##########
src/Consumer.cc:
##########
@@ -192,36 +193,17 @@ struct ConsumerNewInstanceContext {
Napi::Value Consumer::NewInstance(const Napi::CallbackInfo &info, std::shared_ptr<pulsar_client_t> cClient) {
auto deferred = ThreadSafeDeferred::New(info.Env());
auto config = info[0].As<Napi::Object>();
- std::shared_ptr<ConsumerConfig> consumerConfig = std::make_shared<ConsumerConfig>(config, &MessageListener);
+ std::shared_ptr<ConsumerConfig> consumerConfig = std::make_shared<ConsumerConfig>();
+
+ consumerConfig->InitConfig(deferred, config, &MessageListener);
+ if (deferred->IsDone()) {
+ return deferred->Promise();
+ }
const std::string &topic = consumerConfig->GetTopic();
const std::vector<std::string> &topics = consumerConfig->GetTopics();
const std::string &topicsPattern = consumerConfig->GetTopicsPattern();
- if (topic.empty() && topics.size() == 0 && topicsPattern.empty()) {
- deferred->Reject(
- std::string("Topic, topics or topicsPattern is required and must be specified as a string when "
- "creating consumer"));
- return deferred->Promise();
- }
const std::string &subscription = consumerConfig->GetSubscription();
- if (subscription.empty()) {
- deferred->Reject(
- std::string("Subscription is required and must be specified as a string when creating consumer"));
- return deferred->Promise();
- }
- int32_t ackTimeoutMs = consumerConfig->GetAckTimeoutMs();
- if (ackTimeoutMs != 0 && ackTimeoutMs < MIN_ACK_TIMEOUT_MILLIS) {
- std::string msg("Ack timeout should be 0 or greater than or equal to " +
- std::to_string(MIN_ACK_TIMEOUT_MILLIS));
- deferred->Reject(msg);
- return deferred->Promise();
- }
- int32_t nAckRedeliverTimeoutMs = consumerConfig->GetNAckRedeliverTimeoutMs();
- if (nAckRedeliverTimeoutMs < 0) {
- std::string msg("NAck timeout should be greater than or equal to zero");
- deferred->Reject(msg);
- return deferred->Promise();
Review Comment:
Sorry for the late reply. The original code, check settings logic are separated, which will have the following problems:
1. For Some settings, we need to call the C API to determine whether it is successful or not, such as https://github.com/apache/pulsar-client-node/pull/357/files#diff-ca2d7e89de2051db6eacdda40a5abc459724a390b40df60a1e54857caa2ee5f5R321-R325, which we can't do with the original code.
2. After refactoring this way, putting the check settings in a method will make the code clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1440408093
##########
tests/consumer.test.js:
##########
@@ -315,6 +346,47 @@ const Pulsar = require('../index');
consumer.close();
dlqConsumer.close();
});
+
+ test('Batch Receive', async () => {
+ const topicName = 'batch-receive-test-topic';
+ const producer = await client.createProducer({
+ topic: topicName,
+ });
+
+ const consumer = await client.subscribe({
+ topic: topicName,
+ subscription: 'sub1',
+ subscriptionType: 'Shared',
+ batchReceivePolicy: {
+ maxNumMessages: 10,
+ maxNumBytes: -1,
+ timeoutMs: 500,
Review Comment:
Could we add a test case? Sending messages less than 10 to verify timeout works well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1441262640
##########
src/Consumer.cc:
##########
@@ -192,36 +193,17 @@ struct ConsumerNewInstanceContext {
Napi::Value Consumer::NewInstance(const Napi::CallbackInfo &info, std::shared_ptr<pulsar_client_t> cClient) {
auto deferred = ThreadSafeDeferred::New(info.Env());
auto config = info[0].As<Napi::Object>();
- std::shared_ptr<ConsumerConfig> consumerConfig = std::make_shared<ConsumerConfig>(config, &MessageListener);
+ std::shared_ptr<ConsumerConfig> consumerConfig = std::make_shared<ConsumerConfig>();
+
+ consumerConfig->InitConfig(deferred, config, &MessageListener);
+ if (deferred->IsDone()) {
+ return deferred->Promise();
+ }
const std::string &topic = consumerConfig->GetTopic();
const std::vector<std::string> &topics = consumerConfig->GetTopics();
const std::string &topicsPattern = consumerConfig->GetTopicsPattern();
- if (topic.empty() && topics.size() == 0 && topicsPattern.empty()) {
- deferred->Reject(
- std::string("Topic, topics or topicsPattern is required and must be specified as a string when "
- "creating consumer"));
- return deferred->Promise();
- }
const std::string &subscription = consumerConfig->GetSubscription();
- if (subscription.empty()) {
- deferred->Reject(
- std::string("Subscription is required and must be specified as a string when creating consumer"));
- return deferred->Promise();
- }
- int32_t ackTimeoutMs = consumerConfig->GetAckTimeoutMs();
- if (ackTimeoutMs != 0 && ackTimeoutMs < MIN_ACK_TIMEOUT_MILLIS) {
- std::string msg("Ack timeout should be 0 or greater than or equal to " +
- std::to_string(MIN_ACK_TIMEOUT_MILLIS));
- deferred->Reject(msg);
- return deferred->Promise();
- }
- int32_t nAckRedeliverTimeoutMs = consumerConfig->GetNAckRedeliverTimeoutMs();
- if (nAckRedeliverTimeoutMs < 0) {
- std::string msg("NAck timeout should be greater than or equal to zero");
- deferred->Reject(msg);
- return deferred->Promise();
Review Comment:
Why did you remove the checks here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1440406893
##########
src/ConsumerConfig.cc:
##########
@@ -265,6 +291,39 @@ ConsumerConfig::ConsumerConfig(const Napi::Object &consumerConfig, pulsar_messag
}
pulsar_consumer_configuration_set_dlq_policy(this->cConsumerConfig.get(), &dlq_policy);
}
+
+ if (consumerConfig.Has(CFG_BATCH_RECEIVE_POLICY) &&
+ consumerConfig.Get(CFG_BATCH_RECEIVE_POLICY).IsObject()) {
+ Napi::Object propObj = consumerConfig.Get(CFG_BATCH_RECEIVE_POLICY).ToObject();
+ int maxNumMessages = -1;
+ if (propObj.Has(CFG_BATCH_RECEIVE_POLICY_MAX_NUM_MESSAGES) &&
Review Comment:
I found Java client will use the `receiverQueueSize` if `maxNumMessages` is bigger than the `receiverQueueSize`, I'm not sure whether we can add this check in the node client. refer to [here](https://github.com/apache/pulsar/blob/v3.1.2/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java#L148C49-L148C49)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Support batch receive for consumer. [pulsar-client-node]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #357:
URL: https://github.com/apache/pulsar-client-node/pull/357#discussion_r1444116285
##########
src/ConsumerConfig.h:
##########
@@ -21,14 +21,17 @@
#define CONSUMER_CONFIG_H
#include <pulsar/c/consumer_configuration.h>
+#include "ThreadSafeDeferred.h"
#include "MessageListener.h"
#define MIN_ACK_TIMEOUT_MILLIS 10000
class ConsumerConfig {
public:
- ConsumerConfig(const Napi::Object &consumerConfig, pulsar_message_listener messageListener);
+ ConsumerConfig();
~ConsumerConfig();
+ void InitConfig(const std::shared_ptr<ThreadSafeDeferred> deferred, const Napi::Object &consumerConfig,
Review Comment:
I will keep use std::shared_ptr<ThreadSafeDeferred>, because it will create by NAPI method: `auto deferred = ThreadSafeDeferred::New(info.Env());`
##########
src/ConsumerConfig.h:
##########
@@ -21,14 +21,17 @@
#define CONSUMER_CONFIG_H
#include <pulsar/c/consumer_configuration.h>
+#include "ThreadSafeDeferred.h"
#include "MessageListener.h"
#define MIN_ACK_TIMEOUT_MILLIS 10000
class ConsumerConfig {
public:
- ConsumerConfig(const Napi::Object &consumerConfig, pulsar_message_listener messageListener);
+ ConsumerConfig();
~ConsumerConfig();
+ void InitConfig(const std::shared_ptr<ThreadSafeDeferred> deferred, const Napi::Object &consumerConfig,
Review Comment:
I will keep use `std::shared_ptr<ThreadSafeDeferred>`, because it will create by NAPI method: `auto deferred = ThreadSafeDeferred::New(info.Env());`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org