| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| #include <gtest/gtest.h> |
| #include <pulsar/Client.h> |
| |
| #include <thread> |
| |
| #include "HttpHelper.h" |
| #include "PulsarFriend.h" |
| #include "lib/Future.h" |
| #include "lib/Latch.h" |
| #include "lib/LogUtils.h" |
| #include "lib/ProducerImpl.h" |
| #include "lib/Utils.h" |
| DECLARE_LOG_OBJECT() |
| |
| using namespace pulsar; |
| |
| static const std::string serviceUrl = "pulsar://localhost:6650"; |
| static const std::string adminUrl = "http://localhost:8080/"; |
| |
| // See the `maxMessageSize` config in test-conf/standalone-ssl.conf |
| static constexpr size_t maxMessageSize = 1024000; |
| |
| TEST(ProducerTest, producerNotInitialized) { |
| Producer producer; |
| |
| Message msg = MessageBuilder().setContent("test").build(); |
| |
| ASSERT_EQ(ResultProducerNotInitialized, producer.send(msg)); |
| |
| Promise<Result, MessageId> promise; |
| producer.sendAsync(msg, WaitForCallbackValue<MessageId>(promise)); |
| |
| MessageId mi; |
| ASSERT_EQ(ResultProducerNotInitialized, promise.getFuture().get(mi)); |
| |
| ASSERT_EQ(ResultProducerNotInitialized, producer.close()); |
| |
| Promise<bool, Result> promiseClose; |
| producer.closeAsync(WaitForCallback(promiseClose)); |
| |
| Result result; |
| promiseClose.getFuture().get(result); |
| ASSERT_EQ(ResultProducerNotInitialized, result); |
| |
| ASSERT_TRUE(producer.getTopic().empty()); |
| } |
| |
| TEST(ProducerTest, exactlyOnceWithProducerNameSpecified) { |
| Client client(serviceUrl); |
| |
| std::string topicName = "persistent://public/default/exactlyOnceWithProducerNameSpecified"; |
| |
| Producer producer1; |
| ProducerConfiguration producerConfiguration1; |
| producerConfiguration1.setProducerName("p-name-1"); |
| |
| ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfiguration1, producer1)); |
| |
| Producer producer2; |
| ProducerConfiguration producerConfiguration2; |
| producerConfiguration2.setProducerName("p-name-2"); |
| ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfiguration2, producer2)); |
| |
| Producer producer3; |
| Result result = client.createProducer(topicName, producerConfiguration2, producer3); |
| ASSERT_EQ(ResultProducerBusy, result); |
| } |
| |
| TEST(ProducerTest, testSynchronouslySend) { |
| Client client(serviceUrl); |
| const std::string topic = "ProducerTestSynchronouslySend"; |
| |
| Consumer consumer; |
| ASSERT_EQ(ResultOk, client.subscribe(topic, "sub-name", consumer)); |
| |
| Producer producer; |
| ASSERT_EQ(ResultOk, client.createProducer(topic, producer)); |
| MessageId messageId; |
| ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent("hello").build(), messageId)); |
| LOG_INFO("Send message to " << messageId); |
| |
| Message receivedMessage; |
| ASSERT_EQ(ResultOk, consumer.receive(receivedMessage, 3000)); |
| LOG_INFO("Received message from " << receivedMessage.getMessageId()); |
| ASSERT_EQ(receivedMessage.getMessageId(), messageId); |
| ASSERT_EQ(ResultOk, consumer.acknowledge(receivedMessage)); |
| |
| client.close(); |
| } |
| |
| TEST(ProducerTest, testIsConnected) { |
| Client client(serviceUrl); |
| const std::string nonPartitionedTopic = |
| "testProducerIsConnectedNonPartitioned-" + std::to_string(time(nullptr)); |
| const std::string partitionedTopic = |
| "testProducerIsConnectedPartitioned-" + std::to_string(time(nullptr)); |
| |
| Producer producer; |
| ASSERT_FALSE(producer.isConnected()); |
| // ProducerImpl |
| ASSERT_EQ(ResultOk, client.createProducer(nonPartitionedTopic, producer)); |
| ASSERT_TRUE(producer.isConnected()); |
| ASSERT_EQ(ResultOk, producer.close()); |
| ASSERT_FALSE(producer.isConnected()); |
| |
| int res = makePutRequest( |
| adminUrl + "admin/v2/persistent/public/default/" + partitionedTopic + "/partitions", "2"); |
| ASSERT_TRUE(res == 204 || res == 409) << "res: " << res; |
| |
| // PartitionedProducerImpl |
| ASSERT_EQ(ResultOk, client.createProducer(partitionedTopic, producer)); |
| ASSERT_TRUE(producer.isConnected()); |
| ASSERT_EQ(ResultOk, producer.close()); |
| ASSERT_FALSE(producer.isConnected()); |
| |
| client.close(); |
| } |
| |
| TEST(ProducerTest, testSendAsyncAfterCloseAsyncWithLazyProducers) { |
| Client client(serviceUrl); |
| const std::string partitionedTopic = |
| "testProducerIsConnectedPartitioned-" + std::to_string(time(nullptr)); |
| |
| int res = makePutRequest( |
| adminUrl + "admin/v2/persistent/public/default/" + partitionedTopic + "/partitions", "10"); |
| ASSERT_TRUE(res == 204 || res == 409) << "res: " << res; |
| |
| ProducerConfiguration producerConfiguration; |
| producerConfiguration.setLazyStartPartitionedProducers(true); |
| Producer producer; |
| ASSERT_EQ(ResultOk, client.createProducer(partitionedTopic, producerConfiguration, producer)); |
| |
| Message msg = MessageBuilder().setContent("test").build(); |
| |
| Promise<bool, Result> promiseClose; |
| producer.closeAsync(WaitForCallback(promiseClose)); |
| |
| Promise<Result, MessageId> promise; |
| producer.sendAsync(msg, WaitForCallbackValue<MessageId>(promise)); |
| |
| MessageId mi; |
| ASSERT_EQ(ResultAlreadyClosed, promise.getFuture().get(mi)); |
| |
| Result result; |
| promiseClose.getFuture().get(result); |
| ASSERT_EQ(ResultOk, result); |
| } |
| |
| TEST(ProducerTest, testGetNumOfChunks) { |
| ASSERT_EQ(ProducerImpl::getNumOfChunks(11, 5), 3); |
| ASSERT_EQ(ProducerImpl::getNumOfChunks(10, 5), 2); |
| ASSERT_EQ(ProducerImpl::getNumOfChunks(8, 5), 2); |
| ASSERT_EQ(ProducerImpl::getNumOfChunks(4, 5), 1); |
| ASSERT_EQ(ProducerImpl::getNumOfChunks(1, 0), 1); |
| } |
| |
| TEST(ProducerTest, testBacklogQuotasExceeded) { |
| std::string ns = "public/test-backlog-quotas"; |
| std::string topic = ns + "/testBacklogQuotasExceeded" + std::to_string(time(nullptr)); |
| |
| int res = makePutRequest(adminUrl + "admin/v2/persistent/" + topic + "/partitions", "5"); |
| ASSERT_TRUE(res == 204 || res == 409) << "res: " << res; |
| LOG_INFO("Created topic " << topic << " with 5 partitions"); |
| |
| auto setBacklogPolicy = [&ns](const std::string& policy, int limitSize) { |
| const auto body = R"({"policy":")" + policy + R"(","limitSize":)" + std::to_string(limitSize) + "}"; |
| int res = makePostRequest(adminUrl + "admin/v2/namespaces/" + ns + "/backlogQuota", body); |
| LOG_INFO(res << " | Change the backlog policy to: " << body); |
| ASSERT_TRUE(res == 204 || res == 409); |
| std::this_thread::sleep_for(std::chrono::seconds(1)); |
| }; |
| |
| Client client(serviceUrl); |
| |
| // Create a topic with backlog size that is greater than 1024 |
| Consumer consumer; |
| ASSERT_EQ(ResultOk, client.subscribe(topic, "sub", consumer)); // create a cursor |
| Producer producer; |
| |
| const auto partition = topic + "-partition-0"; |
| ASSERT_EQ(ResultOk, client.createProducer(partition, producer)); |
| ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(std::string(1024L, 'a')).build())); |
| ASSERT_EQ(ResultOk, producer.close()); |
| |
| setBacklogPolicy("producer_request_hold", 1024); |
| ASSERT_EQ(ResultProducerBlockedQuotaExceededError, client.createProducer(topic, producer)); |
| ASSERT_EQ(ResultProducerBlockedQuotaExceededError, client.createProducer(partition, producer)); |
| |
| setBacklogPolicy("producer_exception", 1024); |
| ASSERT_EQ(ResultProducerBlockedQuotaExceededException, client.createProducer(topic, producer)); |
| ASSERT_EQ(ResultProducerBlockedQuotaExceededException, client.createProducer(partition, producer)); |
| |
| setBacklogPolicy("consumer_backlog_eviction", 1024); |
| ASSERT_EQ(ResultOk, client.createProducer(topic, producer)); |
| ASSERT_EQ(ResultOk, client.createProducer(partition, producer)); |
| |
| client.close(); |
| } |
| |
| class ProducerTest : public ::testing::TestWithParam<bool> {}; |
| |
| TEST_P(ProducerTest, testMaxMessageSize) { |
| Client client(serviceUrl); |
| |
| const auto topic = std::string("ProducerTest-NoBatchMaxMessageSize-") + |
| (GetParam() ? "batch-" : "-no-batch-") + std::to_string(time(nullptr)); |
| |
| Consumer consumer; |
| ASSERT_EQ(ResultOk, client.subscribe(topic, "sub", consumer)); |
| |
| Producer producer; |
| ProducerConfiguration conf; |
| conf.setBatchingEnabled(GetParam()); |
| ASSERT_EQ(ResultOk, client.createProducer(topic, conf, producer)); |
| |
| std::string msg = std::string(maxMessageSize / 2, 'a'); |
| ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(msg).build())); |
| Message message; |
| ASSERT_EQ(ResultOk, consumer.receive(message)); |
| ASSERT_EQ(msg, message.getDataAsString()); |
| |
| std::string orderKey = std::string(maxMessageSize, 'a'); |
| ASSERT_EQ(ResultMessageTooBig, producer.send(MessageBuilder().setOrderingKey(orderKey).build())); |
| |
| ASSERT_EQ(ResultMessageTooBig, |
| producer.send(MessageBuilder().setContent(std::string(maxMessageSize, 'b')).build())); |
| |
| client.close(); |
| } |
| |
| TEST(ProducerTest, testChunkingMaxMessageSize) { |
| Client client(serviceUrl); |
| |
| const auto topic = std::string("ProducerTest-ChunkingMaxMessageSize-") + std::to_string(time(nullptr)); |
| |
| Consumer consumer; |
| ASSERT_EQ(ResultOk, client.subscribe(topic, "sub", consumer)); |
| |
| Producer producer; |
| ProducerConfiguration conf; |
| conf.setBatchingEnabled(false); |
| conf.setChunkingEnabled(true); |
| ASSERT_EQ(ResultOk, client.createProducer(topic, conf, producer)); |
| |
| std::string orderKey = std::string(maxMessageSize, 'a'); |
| ASSERT_EQ(ResultMessageTooBig, producer.send(MessageBuilder().setOrderingKey(orderKey).build())); |
| |
| std::string msg = std::string(2 * maxMessageSize + 10, 'b'); |
| Message message; |
| ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(msg).build())); |
| ASSERT_EQ(ResultOk, consumer.receive(message)); |
| ASSERT_EQ(msg, message.getDataAsString()); |
| ASSERT_LE(1L, message.getMessageId().entryId()); |
| |
| client.close(); |
| } |
| |
| TEST(ProducerTest, testExclusiveProducer) { |
| Client client(serviceUrl); |
| |
| std::string topicName = |
| "persistent://public/default/testExclusiveProducer" + std::to_string(time(nullptr)); |
| |
| Producer producer1; |
| ProducerConfiguration producerConfiguration1; |
| producerConfiguration1.setProducerName("p-name-1"); |
| producerConfiguration1.setAccessMode(ProducerConfiguration::Exclusive); |
| |
| ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfiguration1, producer1)); |
| |
| Producer producer2; |
| ProducerConfiguration producerConfiguration2; |
| producerConfiguration2.setProducerName("p-name-2"); |
| producerConfiguration2.setAccessMode(ProducerConfiguration::Exclusive); |
| ASSERT_EQ(ResultProducerFenced, client.createProducer(topicName, producerConfiguration2, producer2)); |
| |
| Producer producer3; |
| ProducerConfiguration producerConfiguration3; |
| producerConfiguration3.setProducerName("p-name-3"); |
| ASSERT_EQ(ResultProducerBusy, client.createProducer(topicName, producerConfiguration3, producer3)); |
| } |
| |
| TEST(ProducerTest, testWaitForExclusiveProducer) { |
| Client client(serviceUrl); |
| |
| std::string topicName = |
| "persistent://public/default/testWaitForExclusiveProducer" + std::to_string(time(nullptr)); |
| |
| Producer producer1; |
| ProducerConfiguration producerConfiguration1; |
| producerConfiguration1.setProducerName("p-name-1"); |
| producerConfiguration1.setAccessMode(ProducerConfiguration::Exclusive); |
| |
| ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfiguration1, producer1)); |
| |
| ASSERT_EQ(ResultOk, producer1.send(MessageBuilder().setContent("content").build())); |
| |
| Producer producer2; |
| ProducerConfiguration producerConfiguration2; |
| producerConfiguration2.setProducerName("p-name-2"); |
| producerConfiguration2.setAccessMode(ProducerConfiguration::WaitForExclusive); |
| |
| Latch latch(1); |
| client.createProducerAsync(topicName, producerConfiguration2, |
| [&latch, &producer2](Result res, Producer producer) { |
| ASSERT_EQ(ResultOk, res); |
| producer2 = producer; |
| latch.countdown(); |
| }); |
| |
| // when p1 close, p2 success created. |
| producer1.close(); |
| latch.wait(); |
| ASSERT_EQ(ResultOk, producer2.send(MessageBuilder().setContent("content").build())); |
| |
| producer2.close(); |
| } |
| |
| TEST(ProducerTest, testExclusiveWithFencingProducer) { |
| Client client(serviceUrl); |
| |
| std::string topicName = |
| "persistent://public/default/testExclusiveWithFencingProducer" + std::to_string(time(nullptr)); |
| |
| Producer producer1; |
| ProducerConfiguration producerConfiguration1; |
| producerConfiguration1.setProducerName("p-name-1"); |
| producerConfiguration1.setAccessMode(ProducerConfiguration::Exclusive); |
| |
| ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfiguration1, producer1)); |
| producer1.send(MessageBuilder().setContent("content").build()); |
| |
| Producer producer2; |
| ProducerConfiguration producerConfiguration2; |
| producerConfiguration2.setProducerName("p-name-2"); |
| producerConfiguration2.setAccessMode(ProducerConfiguration::WaitForExclusive); |
| |
| Latch latch(1); |
| client.createProducerAsync(topicName, producerConfiguration2, |
| [&latch, &producer2](Result res, Producer producer) { |
| // producer2 will be fenced |
| ASSERT_EQ(ResultProducerFenced, res); |
| latch.countdown(); |
| producer2 = producer; |
| }); |
| // wait for all the Producers to be enqueued in order to prevent races |
| sleep(1); |
| |
| // producer3 will create success. |
| Producer producer3; |
| ProducerConfiguration producerConfiguration3; |
| producerConfiguration3.setProducerName("p-name-3"); |
| producerConfiguration3.setAccessMode(ProducerConfiguration::ExclusiveWithFencing); |
| ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfiguration3, producer3)); |
| ASSERT_EQ(ResultOk, producer3.send(MessageBuilder().setContent("content").build())); |
| |
| latch.wait(); |
| // producer1 will be fenced |
| ASSERT_EQ(ResultProducerFenced, producer1.send(MessageBuilder().setContent("content").build())); |
| |
| // Again create producer4 with WaitForExclusive |
| Producer producer4; |
| ProducerConfiguration producerConfiguration4; |
| producerConfiguration2.setProducerName("p-name-4"); |
| producerConfiguration2.setAccessMode(ProducerConfiguration::WaitForExclusive); |
| Latch latch2(1); |
| client.createProducerAsync(topicName, producerConfiguration2, |
| [&latch2, &producer4](Result res, Producer producer) { |
| // producer4 will be success |
| ASSERT_EQ(ResultOk, res); |
| producer4 = producer; |
| latch2.countdown(); |
| }); |
| ASSERT_EQ(ResultProducerNotInitialized, producer4.send(MessageBuilder().setContent("content").build())); |
| |
| // When producer3 is close, producer4 will be create success |
| producer3.close(); |
| latch2.wait(); |
| ASSERT_EQ(ResultOk, producer4.send(MessageBuilder().setContent("content").build())); |
| |
| client.close(); |
| } |
| |
| TEST_P(ProducerTest, testFlushNoBatch) { |
| Client client(serviceUrl); |
| |
| auto partitioned = GetParam(); |
| const auto topicName = std::string("testFlushNoBatch") + |
| (partitioned ? "partitioned-" : "-no-partitioned-") + |
| std::to_string(time(nullptr)); |
| |
| if (partitioned) { |
| // call admin api to make it partitioned |
| std::string url = adminUrl + "admin/v2/persistent/public/default/" + topicName + "/partitions"; |
| int res = makePutRequest(url, "5"); |
| LOG_INFO("res = " << res); |
| ASSERT_FALSE(res != 204 && res != 409); |
| } |
| |
| ProducerConfiguration producerConfiguration; |
| producerConfiguration.setBatchingEnabled(false); |
| |
| Producer producer; |
| ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfiguration, producer)); |
| |
| std::atomic_int needCallBack(100); |
| auto cb = [&needCallBack](Result code, const MessageId& msgId) { |
| ASSERT_EQ(code, ResultOk); |
| needCallBack.fetch_sub(1); |
| }; |
| |
| for (int i = 0; i < 100; ++i) { |
| Message msg = MessageBuilder().setContent("content").build(); |
| producer.sendAsync(msg, cb); |
| } |
| |
| producer.flush(); |
| ASSERT_EQ(needCallBack.load(), 0); |
| producer.close(); |
| |
| client.close(); |
| } |
| |
| TEST(ProducerTest, testCloseSubProducerWhenFail) { |
| Client client(serviceUrl); |
| |
| std::string ns = "test-close-sub-producer-when-fail"; |
| std::string localName = std::string("testCloseSubProducerWhenFail") + std::to_string(time(nullptr)); |
| std::string topicName = "persistent://public/" + ns + '/' + localName; |
| const int maxProducersPerTopic = 10; |
| const int partitionNum = 5; |
| |
| // call admin api to create namespace with max prodcuer limit |
| std::string url = adminUrl + "admin/v2/namespaces/public/" + ns; |
| int res = |
| makePutRequest(url, "{\"max_producers_per_topic\": " + std::to_string(maxProducersPerTopic) + "}"); |
| ASSERT_TRUE(res == 204 || res == 409) << "res:" << res; |
| |
| // call admin api to create partitioned topic |
| res = makePutRequest(adminUrl + "admin/v2/persistent/public/" + ns + "/" + localName + "/partitions", |
| std::to_string(partitionNum)); |
| ASSERT_TRUE(res == 204 || res == 409) << "res: " << res; |
| |
| ProducerConfiguration producerConfiguration; |
| producerConfiguration.setBatchingEnabled(false); |
| |
| // create producers for partition-0 up to max producer limit |
| std::vector<Producer> producers; |
| for (int i = 0; i < maxProducersPerTopic; ++i) { |
| Producer producer; |
| ASSERT_EQ(ResultOk, |
| client.createProducer(topicName + "-partition-0", producerConfiguration, producer)); |
| producers.push_back(producer); |
| } |
| |
| // create partitioned producer, should fail because partition-0 already reach max producer limit |
| for (int i = 0; i < maxProducersPerTopic; ++i) { |
| Producer producer; |
| ASSERT_EQ(ResultProducerBusy, client.createProducer(topicName, producer)); |
| } |
| |
| std::this_thread::sleep_for(std::chrono::seconds(1)); |
| |
| // create producer for partition-1, should succeed |
| Producer producer; |
| ASSERT_EQ(ResultOk, client.createProducer(topicName + "-partition-1", producerConfiguration, producer)); |
| producers.push_back(producer); |
| |
| for (auto& producer : producers) { |
| producer.close(); |
| } |
| client.close(); |
| } |
| |
| TEST(ProducerTest, testCloseProducerBeforeCreated) { |
| Client client(serviceUrl); |
| |
| std::string ns = "test-close-producer-before-created"; |
| std::string localName = std::string("testCloseProducerBeforeCreated") + std::to_string(time(nullptr)); |
| std::string topicName = "persistent://public/" + ns + '/' + localName; |
| const int maxProducersPerTopic = 10; |
| const int partitionNum = 5; |
| |
| // call admin api to create namespace with max prodcuer limit |
| std::string url = adminUrl + "admin/v2/namespaces/public/" + ns; |
| int res = |
| makePutRequest(url, "{\"max_producers_per_topic\": " + std::to_string(maxProducersPerTopic) + "}"); |
| ASSERT_TRUE(res == 204 || res == 409) << "res:" << res; |
| |
| // call admin api to create partitioned topic |
| res = makePutRequest(adminUrl + "admin/v2/persistent/public/" + ns + "/" + localName + "/partitions", |
| std::to_string(partitionNum)); |
| ASSERT_TRUE(res == 204 || res == 409) << "res: " << res; |
| |
| ProducerConfiguration producerConfiguration; |
| producerConfiguration.setLazyStartPartitionedProducers(true); |
| producerConfiguration.setPartitionsRoutingMode(ProducerConfiguration::RoundRobinDistribution); |
| producerConfiguration.setBatchingEnabled(false); |
| |
| Message msg = MessageBuilder().setContent("test").build(); |
| for (int i = 0; i < maxProducersPerTopic * 100; ++i) { |
| Producer producer; |
| ASSERT_EQ(ResultOk, client.createProducer(topicName, producerConfiguration, producer)); |
| // trigger lazy producer creation |
| for (int j = 0; j < partitionNum; ++j) { |
| producer.sendAsync(msg, [](pulsar::Result, const pulsar::MessageId&) {}); |
| } |
| producer.close(); |
| } |
| |
| Producer producer; |
| ASSERT_EQ(ResultOk, client.createProducer(topicName, {}, producer)); |
| producer.close(); |
| |
| client.close(); |
| } |
| |
| TEST(ProducerTest, testNoDeadlockWhenClosingPartitionedProducerAfterPartitionsUpdate) { |
| const std::string topic = "public/default/partition-test" + std::to_string(time(nullptr)); |
| std::string topicOperateUrl = adminUrl + "admin/v2/persistent/" + topic + "/partitions"; |
| |
| int res = makePutRequest(topicOperateUrl, "2"); |
| ASSERT_TRUE(res == 204 || res == 409) << "res: " << res; |
| |
| ClientConfiguration clientConf; |
| clientConf.setPartititionsUpdateInterval(1); |
| Client client(serviceUrl, clientConf); |
| ProducerConfiguration conf; |
| Producer producer; |
| client.createProducer(topic, conf, producer); |
| PartitionedProducerImpl& partitionedProducer = PulsarFriend::getPartitionedProducerImpl(producer); |
| |
| // TODO: Replace by producer interceptor to reproduce the issue then we can remove |
| // PulsarFriend::updatePartitions |
| PulsarFriend::updatePartitions(partitionedProducer, 3); |
| |
| producer.close(); |
| client.close(); |
| } |
| |
| INSTANTIATE_TEST_CASE_P(Pulsar, ProducerTest, ::testing::Values(true, false)); |
