You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/03/30 14:20:25 UTC
[pulsar] branch branch-2.7 updated (eddc5f0 -> a187eb7)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from eddc5f0 [api][pulsar-client]Add get version command for pulsar rest api, pulsar-admin, pulsar-client (#9975)
new adc04f6 [Pulsar-sql]Using pulsar SQL query messages will appear `NoSuchLedger… (#9910)
new 668c692 [C++] Fix paused zero queue consumer still pre-fetches messages (#10036)
new 5d00904 bugfix for wrong timeunit in updating lastLedgerCreationInitiationTimestamp (#10049)
new a23f31c Fix possible name mismatch bugs when build wheel files within docker (#10051)
new 8ed886e Upgrade Netty version to 4.1.60.final (#10073)
new 0eb7311 Fix checkstyle issue
new a187eb7 [ISSUE 9783][pulsar-client] Allow pulsar client receive external timer (#9802)
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
distribution/server/src/assemble/LICENSE.bin.txt | 36 +++----
.../mledger/impl/ManagedLedgerFactoryImpl.java | 2 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 2 +-
.../ManagedLedgerFactoryChangeLedgerPathTest.java | 36 +++++++
pom.xml | 4 +-
.../docker/build-wheel-file-within-docker.sh | 2 +-
pulsar-client-cpp/lib/ConsumerImpl.cc | 57 ++++-------
pulsar-client-cpp/lib/ConsumerImpl.h | 12 +--
pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc | 2 +-
pulsar-client-cpp/lib/PartitionedConsumerImpl.cc | 2 +-
pulsar-client-cpp/tests/ZeroQueueSizeTest.cc | 107 +++++++++++++++++++++
.../pulsar/client/impl/PulsarClientImpl.java | 19 +++-
.../pulsar/client/impl/PulsarClientImplTest.java | 32 ++++++
.../pulsar/proxy/server/ProxyConnection.java | 4 +-
.../apache/pulsar/proxy/server/ProxyService.java | 11 +++
pulsar-sql/presto-distribution/LICENSE | 28 +++---
.../pulsar/sql/presto/PulsarConnectorCache.java | 5 +-
17 files changed, 273 insertions(+), 88 deletions(-)
[pulsar] 02/07: [C++] Fix paused zero queue consumer still
pre-fetches messages (#10036)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 668c692782ff25ecfb953ecd7a1e148140093c29
Author: Yunze Xu <xy...@163.com>
AuthorDate: Sat Mar 27 06:48:18 2021 +0800
[C++] Fix paused zero queue consumer still pre-fetches messages (#10036)
* Fix zero queue consumer pre-fetches messages after paused
* Remove unused logs and checks
* Fix pulsar image build failure
* Fix AtomicHelper functions
* Use fetch_add as a substitute of addAndGet
(cherry picked from commit fb6b4ea04dda003f35007c180a1edc22ae619591)
---
pulsar-client-cpp/lib/ConsumerImpl.cc | 57 +++++-------
pulsar-client-cpp/lib/ConsumerImpl.h | 12 +--
pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc | 2 +-
pulsar-client-cpp/lib/PartitionedConsumerImpl.cc | 2 +-
pulsar-client-cpp/tests/ZeroQueueSizeTest.cc | 107 +++++++++++++++++++++++
5 files changed, 136 insertions(+), 44 deletions(-)
diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc
index d75d018..a391ae5 100644
--- a/pulsar-client-cpp/lib/ConsumerImpl.cc
+++ b/pulsar-client-cpp/lib/ConsumerImpl.cc
@@ -54,7 +54,8 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic,
// This is the initial capacity of the queue
incomingMessages_(std::max(config_.getReceiverQueueSize(), 1)),
pendingReceives_(),
- availablePermits_(conf.getReceiverQueueSize()),
+ availablePermits_(0),
+ receiverQueueRefillThreshold_(config_.getReceiverQueueSize() / 2),
consumerId_(client->newConsumerId()),
consumerName_(config_.getConsumerName()),
partitionIndex_(-1),
@@ -202,9 +203,12 @@ void ConsumerImpl::connectionFailed(Result result) {
}
}
-void ConsumerImpl::receiveMessages(const ClientConnectionPtr& cnx, unsigned int count) {
- SharedBuffer cmd = Commands::newFlow(consumerId_, count);
- cnx->sendCommand(cmd);
+void ConsumerImpl::sendFlowPermitsToBroker(const ClientConnectionPtr& cnx, int numMessages) {
+ if (cnx) {
+ LOG_DEBUG(getName() << "Send more permits: " << numMessages);
+ SharedBuffer cmd = Commands::newFlow(consumerId_, static_cast<unsigned int>(numMessages));
+ cnx->sendCommand(cmd);
+ }
}
void ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result result) {
@@ -223,7 +227,7 @@ void ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result r
backoff_.reset();
// Complicated logic since we don't have a isLocked() function for mutex
if (waitingForZeroQueueSizeMessage) {
- receiveMessages(cnx, 1);
+ sendFlowPermitsToBroker(cnx, 1);
}
availablePermits_ = 0;
}
@@ -231,9 +235,9 @@ void ConsumerImpl::handleCreateConsumer(const ClientConnectionPtr& cnx, Result r
LOG_DEBUG(getName() << "Send initial flow permits: " << config_.getReceiverQueueSize());
if (consumerTopicType_ == NonPartitioned || !firstTime) {
if (config_.getReceiverQueueSize() != 0) {
- receiveMessages(cnx, config_.getReceiverQueueSize());
+ sendFlowPermitsToBroker(cnx, config_.getReceiverQueueSize());
} else if (messageListener_) {
- receiveMessages(cnx, 1);
+ sendFlowPermitsToBroker(cnx, 1);
}
}
consumerCreatedPromise_.setValue(shared_from_this());
@@ -380,11 +384,9 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::
}
if (messageListener_) {
- Lock lock(messageListenerMutex_);
if (!messageListenerRunning_) {
return;
}
- lock.unlock();
// Trigger message listener callback in a separate thread
while (numOfMessageReceived--) {
listenerExecutor_->postWork(std::bind(&ConsumerImpl::internalListener, shared_from_this()));
@@ -556,11 +558,9 @@ void ConsumerImpl::discardCorruptedMessage(const ClientConnectionPtr& cnx,
}
void ConsumerImpl::internalListener() {
- Lock lock(messageListenerMutex_);
if (!messageListenerRunning_) {
return;
}
- lock.unlock();
Message msg;
if (!incomingMessages_.pop(msg, std::chrono::milliseconds(0))) {
// This will only happen when the connection got reset and we cleared the queue
@@ -596,10 +596,7 @@ Result ConsumerImpl::fetchSingleMessageFromBroker(Message& msg) {
waitingForZeroQueueSizeMessage = true;
localLock.unlock();
- if (currentCnx) {
- LOG_DEBUG(getName() << "Send more permits: " << 1);
- receiveMessages(currentCnx, 1);
- }
+ sendFlowPermitsToBroker(currentCnx, 1);
while (true) {
incomingMessages_.pop(msg);
@@ -646,11 +643,7 @@ void ConsumerImpl::receiveAsync(ReceiveCallback& callback) {
lock.unlock();
if (config_.getReceiverQueueSize() == 0) {
- ClientConnectionPtr currentCnx = getCnx().lock();
- if (currentCnx) {
- LOG_DEBUG(getName() << "Send more permits: " << 1);
- receiveMessages(currentCnx, 1);
- }
+ sendFlowPermitsToBroker(getCnx().lock(), 1);
}
}
}
@@ -752,20 +745,13 @@ Optional<MessageId> ConsumerImpl::clearReceiveQueue() {
}
}
-void ConsumerImpl::increaseAvailablePermits(const ClientConnectionPtr& currentCnx, int numberOfPermits) {
- int additionalPermits = 0;
+void ConsumerImpl::increaseAvailablePermits(const ClientConnectionPtr& currentCnx, int delta) {
+ int newAvailablePermits = availablePermits_.fetch_add(delta) + delta;
- availablePermits_ += numberOfPermits;
- if (availablePermits_ >= config_.getReceiverQueueSize() / 2) {
- additionalPermits = availablePermits_;
- availablePermits_ = 0;
- }
- if (additionalPermits > 0) {
- if (currentCnx) {
- LOG_DEBUG(getName() << "Send more permits: " << additionalPermits);
- receiveMessages(currentCnx, additionalPermits);
- } else {
- LOG_DEBUG(getName() << "Connection is not ready, Unable to send flow Command");
+ while (newAvailablePermits >= receiverQueueRefillThreshold_ && messageListenerRunning_) {
+ if (availablePermits_.compare_exchange_weak(newAvailablePermits, 0)) {
+ sendFlowPermitsToBroker(currentCnx, newAvailablePermits);
+ break;
}
}
}
@@ -972,7 +958,6 @@ Result ConsumerImpl::pauseMessageListener() {
if (!messageListener_) {
return ResultInvalidConfiguration;
}
- Lock lock(messageListenerMutex_);
messageListenerRunning_ = false;
return ResultOk;
}
@@ -982,19 +967,19 @@ Result ConsumerImpl::resumeMessageListener() {
return ResultInvalidConfiguration;
}
- Lock lock(messageListenerMutex_);
if (messageListenerRunning_) {
// Not paused
return ResultOk;
}
messageListenerRunning_ = true;
const size_t count = incomingMessages_.size();
- lock.unlock();
for (size_t i = 0; i < count; i++) {
// Trigger message listener callback in a separate thread
listenerExecutor_->postWork(std::bind(&ConsumerImpl::internalListener, shared_from_this()));
}
+ // Check current permits and determine whether to send FLOW command
+ this->increaseAvailablePermits(getCnx().lock(), 0);
return ResultOk;
}
diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h
index 1029e2f..46957ca 100644
--- a/pulsar-client-cpp/lib/ConsumerImpl.h
+++ b/pulsar-client-cpp/lib/ConsumerImpl.h
@@ -43,6 +43,7 @@
#include <lib/stats/ConsumerStatsImpl.h>
#include <lib/stats/ConsumerStatsDisabled.h>
#include <queue>
+#include <atomic>
using namespace pulsar;
@@ -73,11 +74,10 @@ class ConsumerImpl : public ConsumerImplBase,
~ConsumerImpl();
void setPartitionIndex(int partitionIndex);
int getPartitionIndex();
- void receiveMessages(const ClientConnectionPtr& cnx, unsigned int count);
+ void sendFlowPermitsToBroker(const ClientConnectionPtr& cnx, int numMessages);
uint64_t getConsumerId();
void messageReceived(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg,
bool& isChecksumValid, proto::MessageMetadata& msgMetadata, SharedBuffer& payload);
- int incrementAndGetPermits(uint64_t cnxSequenceId);
void messageProcessed(Message& msg);
inline proto::CommandSubscribe_SubType getSubType();
inline proto::CommandSubscribe_InitialPosition getInitialPosition();
@@ -150,7 +150,7 @@ class ConsumerImpl : public ConsumerImplBase,
const proto::MessageMetadata& metadata, SharedBuffer& payload);
void discardCorruptedMessage(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageId,
proto::CommandAck::ValidationError validationError);
- void increaseAvailablePermits(const ClientConnectionPtr& currentCnx, int numberOfPermits = 1);
+ void increaseAvailablePermits(const ClientConnectionPtr& currentCnx, int delta = 1);
void drainIncomingMessageQueue(size_t count);
uint32_t receiveIndividualMessagesFromBatch(const ClientConnectionPtr& cnx, Message& batchedMessage,
int redeliveryCount);
@@ -185,14 +185,14 @@ class ConsumerImpl : public ConsumerImplBase,
Optional<MessageId> lastDequedMessage_;
UnboundedBlockingQueue<Message> incomingMessages_;
std::queue<ReceiveCallback> pendingReceives_;
- int availablePermits_;
+ std::atomic_int availablePermits_;
+ const int receiverQueueRefillThreshold_;
uint64_t consumerId_;
std::string consumerName_;
std::string consumerStr_;
int32_t partitionIndex_;
Promise<Result, ConsumerImplBaseWeakPtr> consumerCreatedPromise_;
- bool messageListenerRunning_;
- std::mutex messageListenerMutex_;
+ std::atomic_bool messageListenerRunning_;
CompressionCodecProvider compressionCodecProvider_;
UnAckedMessageTrackerPtr unAckedMessageTrackerPtr_;
BatchAcknowledgementTracker batchAcknowledgementTracker_;
diff --git a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc
index b62083a..46660e7 100644
--- a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc
+++ b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc
@@ -613,7 +613,7 @@ void MultiTopicsConsumerImpl::receiveMessages() {
for (ConsumerMap::const_iterator consumer = consumers_.begin(); consumer != consumers_.end();
consumer++) {
ConsumerImplPtr consumerPtr = consumer->second;
- consumerPtr->receiveMessages(consumerPtr->getCnx().lock(), conf_.getReceiverQueueSize());
+ consumerPtr->sendFlowPermitsToBroker(consumerPtr->getCnx().lock(), conf_.getReceiverQueueSize());
LOG_DEBUG("Sending FLOW command for consumer - " << consumerPtr->getConsumerId());
}
}
diff --git a/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc b/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc
index 3a43a0b..bf12362 100644
--- a/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc
+++ b/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc
@@ -452,7 +452,7 @@ void PartitionedConsumerImpl::internalListener(Consumer consumer) {
void PartitionedConsumerImpl::receiveMessages() {
for (ConsumerList::const_iterator i = consumers_.begin(); i != consumers_.end(); i++) {
ConsumerImplPtr consumer = *i;
- consumer->receiveMessages(consumer->getCnx().lock(), conf_.getReceiverQueueSize());
+ consumer->sendFlowPermitsToBroker(consumer->getCnx().lock(), conf_.getReceiverQueueSize());
LOG_DEBUG("Sending FLOW command for consumer - " << consumer->getConsumerId());
}
}
diff --git a/pulsar-client-cpp/tests/ZeroQueueSizeTest.cc b/pulsar-client-cpp/tests/ZeroQueueSizeTest.cc
index 53b6c9d..6d6c933 100644
--- a/pulsar-client-cpp/tests/ZeroQueueSizeTest.cc
+++ b/pulsar-client-cpp/tests/ZeroQueueSizeTest.cc
@@ -20,7 +20,12 @@
#include <pulsar/Client.h>
#include <lib/Latch.h>
#include "ConsumerTest.h"
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
#include <functional>
+#include <memory>
+#include <mutex>
DECLARE_LOG_OBJECT()
@@ -116,3 +121,105 @@ TEST(ZeroQueueSizeTest, testMessageListener) {
producer.close();
client.close();
}
+
+static ConsumerConfiguration zeroQueueSharedConsumerConf(
+ const std::string& name, std::function<void(Consumer, const Message&)> callback) {
+ ConsumerConfiguration conf;
+ conf.setConsumerType(ConsumerShared);
+ conf.setReceiverQueueSize(0);
+ conf.setSubscriptionInitialPosition(InitialPositionEarliest);
+ conf.setMessageListener([name, callback](Consumer consumer, const Message& msg) {
+ LOG_INFO(name << " received " << msg.getDataAsString() << " from " << msg.getMessageId());
+ callback(consumer, msg);
+ });
+ return conf;
+}
+
+class IntVector {
+ public:
+ size_t add(int i) {
+ std::lock_guard<std::mutex> lock(mutex_);
+ data_.emplace_back(i);
+ return data_.size();
+ }
+
+ std::vector<int> data() const {
+ std::lock_guard<std::mutex> lock(mutex_);
+ return data_;
+ }
+
+ private:
+ std::vector<int> data_;
+ mutable std::mutex mutex_;
+};
+
+TEST(ZeroQueueSizeTest, testPauseResume) {
+ Client client(lookupUrl);
+ const auto topic = "ZeroQueueSizeTestPauseListener-" + std::to_string(time(nullptr));
+ const auto subscription = "my-sub";
+
+ auto intToMessage = [](int i) { return MessageBuilder().setContent(std::to_string(i)).build(); };
+ auto messageToInt = [](const Message& msg) { return std::stoi(msg.getDataAsString()); };
+
+ // 1. Produce 10 messages
+ Producer producer;
+ const auto producerConf = ProducerConfiguration().setBatchingEnabled(false);
+ ASSERT_EQ(ResultOk, client.createProducer(topic, producerConf, producer));
+ for (int i = 0; i < 10; i++) {
+ MessageId id;
+ ASSERT_EQ(ResultOk, producer.send(intToMessage(i), id));
+ LOG_INFO("Send " << i << " to " << id);
+ }
+
+ // 2. consumer-1 receives 1 message and pause
+ std::mutex mtx;
+ std::condition_variable condConsumer1FirstMessage;
+ std::condition_variable condConsumer1Completed;
+ IntVector messages1;
+ const auto conf1 = zeroQueueSharedConsumerConf("consumer-1", [&](Consumer consumer, const Message& msg) {
+ const auto numReceived = messages1.add(messageToInt(msg));
+ if (numReceived == 1) {
+ ASSERT_EQ(ResultOk, consumer.pauseMessageListener());
+ condConsumer1FirstMessage.notify_all();
+ } else if (numReceived == 5) {
+ ASSERT_EQ(ResultOk, consumer.pauseMessageListener());
+ condConsumer1Completed.notify_all();
+ }
+ });
+ Consumer consumer1;
+ ASSERT_EQ(ResultOk, client.subscribe(topic, subscription, conf1, consumer1));
+ {
+ std::unique_lock<std::mutex> lock(mtx);
+ ASSERT_EQ(condConsumer1FirstMessage.wait_for(lock, std::chrono::seconds(3)),
+ std::cv_status::no_timeout);
+ ASSERT_EQ(messages1.data(), (std::vector<int>{0}));
+ }
+
+ // 3. consumer-2 receives 5 messages and pause
+ std::condition_variable condConsumer2Completed;
+ IntVector messages2;
+ const auto conf2 = zeroQueueSharedConsumerConf("consumer-2", [&](Consumer consumer, const Message& msg) {
+ const int numReceived = messages2.add(messageToInt(msg));
+ if (numReceived == 5) {
+ ASSERT_EQ(ResultOk, consumer.pauseMessageListener());
+ condConsumer2Completed.notify_all();
+ }
+ });
+ Consumer consumer2;
+ ASSERT_EQ(ResultOk, client.subscribe(topic, subscription, conf2, consumer2));
+ {
+ std::unique_lock<std::mutex> lock(mtx);
+ ASSERT_EQ(condConsumer2Completed.wait_for(lock, std::chrono::seconds(3)), std::cv_status::no_timeout);
+ ASSERT_EQ(messages2.data(), (std::vector<int>{1, 2, 3, 4, 5}));
+ }
+
+ // 4. consumer-1 resumes listening, and receives last 4 messages
+ ASSERT_EQ(ResultOk, consumer1.resumeMessageListener());
+ {
+ std::unique_lock<std::mutex> lock(mtx);
+ ASSERT_EQ(condConsumer1Completed.wait_for(lock, std::chrono::seconds(3)), std::cv_status::no_timeout);
+ ASSERT_EQ(messages1.data(), (std::vector<int>{0, 6, 7, 8, 9}));
+ }
+
+ client.close();
+}
[pulsar] 07/07: [ISSUE 9783][pulsar-client] Allow pulsar client
receive external timer (#9802)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a187eb72b5dbc429dac1cb313a64b736622ed14f
Author: linlinnn <li...@163.com>
AuthorDate: Mon Mar 29 11:19:42 2021 +0800
[ISSUE 9783][pulsar-client] Allow pulsar client receive external timer (#9802)
Fixed #9783
Allow pulsar client to receive external timer instance
Add new constructor to provide an external timer, and share timer in pulsar proxy
- [x] Make sure that the change passes the CI checks.
(cherry picked from commit af6eabae335ed009dc8a54408ffda6c77348db7f)
---
.../pulsar/client/impl/PulsarClientImpl.java | 19 +++++++++++--
.../pulsar/client/impl/PulsarClientImplTest.java | 32 ++++++++++++++++++++++
.../pulsar/proxy/server/ProxyConnection.java | 4 +--
.../apache/pulsar/proxy/server/ProxyService.java | 11 ++++++++
.../pulsar/sql/presto/PulsarConnectorCache.java | 2 +-
5 files changed, 62 insertions(+), 6 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index 43b8acb..7aa554b 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -94,6 +94,7 @@ public class PulsarClientImpl implements PulsarClient {
private LookupService lookup;
private final ConnectionPool cnxPool;
private final Timer timer;
+ private boolean needStopTimer;
private final ExecutorProvider externalExecutorProvider;
private final ExecutorProvider internalExecutorService;
@@ -130,11 +131,16 @@ public class PulsarClientImpl implements PulsarClient {
}
public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup) throws PulsarClientException {
- this(conf, eventLoopGroup, new ConnectionPool(conf, eventLoopGroup));
+ this(conf, eventLoopGroup, new ConnectionPool(conf, eventLoopGroup), null);
}
public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool cnxPool)
throws PulsarClientException {
+ this(conf, eventLoopGroup, cnxPool, null);
+ }
+
+ public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool cnxPool, Timer timer)
+ throws PulsarClientException {
if (conf == null || isBlank(conf.getServiceUrl()) || eventLoopGroup == null) {
throw new PulsarClientException.InvalidConfigurationException("Invalid client configuration");
}
@@ -151,7 +157,12 @@ public class PulsarClientImpl implements PulsarClient {
} else {
lookup = new BinaryProtoLookupService(this, conf.getServiceUrl(), conf.getListenerName(), conf.isUseTls(), externalExecutorProvider.getExecutor());
}
- timer = new HashedWheelTimer(getThreadFactory("pulsar-timer"), 1, TimeUnit.MILLISECONDS);
+ if (timer == null) {
+ this.timer = new HashedWheelTimer(getThreadFactory("pulsar-timer"), 1, TimeUnit.MILLISECONDS);
+ needStopTimer = true;
+ } else {
+ this.timer = timer;
+ }
producers = Collections.newSetFromMap(new ConcurrentHashMap<>());
consumers = Collections.newSetFromMap(new ConcurrentHashMap<>());
@@ -601,7 +612,9 @@ public class PulsarClientImpl implements PulsarClient {
try {
lookup.close();
cnxPool.close();
- timer.stop();
+ if (needStopTimer) {
+ timer.stop();
+ }
externalExecutorProvider.shutdownNow();
internalExecutorService.shutdownNow();
conf.getAuthentication().close();
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java
index f7df415..d712c26 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PulsarClientImplTest.java
@@ -22,6 +22,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertSame;
@@ -33,8 +34,10 @@ import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoopGroup;
+import io.netty.util.HashedWheelTimer;
import io.netty.util.concurrent.DefaultThreadFactory;
+import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
@@ -54,6 +57,7 @@ import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
import org.mockito.internal.util.reflection.FieldSetter;
+import org.mockito.Mockito;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -146,4 +150,32 @@ public class PulsarClientImplTest {
assertSame(consumer.getState(), HandlerState.State.Closed));
}
+ @Test
+ public void testInitializeWithoutTimer() throws Exception {
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setServiceUrl("pulsar://localhost:6650");
+ PulsarClientImpl client = new PulsarClientImpl(conf);
+
+ HashedWheelTimer timer = mock(HashedWheelTimer.class);
+ Field field = client.getClass().getDeclaredField("timer");
+ field.setAccessible(true);
+ field.set(client, timer);
+
+ client.shutdown();
+ verify(timer).stop();
+ }
+
+ @Test
+ public void testInitializeWithTimer() throws PulsarClientException {
+ ClientConfigurationData conf = new ClientConfigurationData();
+ EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, new DefaultThreadFactory("test"));
+ ConnectionPool pool = Mockito.spy(new ConnectionPool(conf, eventLoop));
+ conf.setServiceUrl("pulsar://localhost:6650");
+
+ HashedWheelTimer timer = new HashedWheelTimer();
+ PulsarClientImpl client = new PulsarClientImpl(conf, eventLoop, pool, timer);
+
+ client.shutdown();
+ client.timer().stop();
+ }
}
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java
index 63e652c..a483ee3 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java
@@ -299,7 +299,7 @@ public class ProxyConnection extends PulsarHandler implements FutureListener<Voi
if (!service.getConfiguration().isAuthenticationEnabled()) {
this.client = new PulsarClientImpl(clientConf, service.getWorkerGroup(),
new ProxyConnectionPool(clientConf, service.getWorkerGroup(),
- () -> new ClientCnx(clientConf, service.getWorkerGroup(), protocolVersion)));
+ () -> new ClientCnx(clientConf, service.getWorkerGroup(), protocolVersion)), service.getTimer());
completeConnect();
return;
@@ -436,7 +436,7 @@ public class ProxyConnection extends PulsarHandler implements FutureListener<Voi
final String clientAuthMethod, final int protocolVersion) throws PulsarClientException {
return new PulsarClientImpl(clientConf, service.getWorkerGroup(),
new ProxyConnectionPool(clientConf, service.getWorkerGroup(), () -> new ProxyClientCnx(clientConf,
- service.getWorkerGroup(), clientAuthRole, clientAuthData, clientAuthMethod, protocolVersion)));
+ service.getWorkerGroup(), clientAuthRole, clientAuthData, clientAuthMethod, protocolVersion)), service.getTimer());
}
private static int getProtocolVersionToAdvertise(CommandConnect connect) {
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
index 58ba3cf..c2e157c 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java
@@ -27,6 +27,8 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.util.concurrent.DefaultThreadFactory;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timer;
import io.prometheus.client.Counter;
import io.prometheus.client.Gauge;
import lombok.Getter;
@@ -67,6 +69,7 @@ import com.google.common.collect.Sets;
public class ProxyService implements Closeable {
private final ProxyConfiguration proxyConfig;
+ private final Timer timer;
private String serviceUrl;
private String serviceUrlTls;
private ConfigurationCacheService configurationCacheService;
@@ -124,6 +127,7 @@ public class ProxyService implements Closeable {
AuthenticationService authenticationService) throws IOException {
checkNotNull(proxyConfig);
this.proxyConfig = proxyConfig;
+ this.timer = new HashedWheelTimer(new DefaultThreadFactory("pulsar-timer", Thread.currentThread().isDaemon()), 1, TimeUnit.MILLISECONDS);
this.clientCnxs = Sets.newConcurrentHashSet();
this.topicStats = Maps.newConcurrentMap();
@@ -243,6 +247,9 @@ public class ProxyService implements Closeable {
acceptorGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
+ if (timer != null) {
+ timer.stop();
+ }
}
public String getServiceUrl() {
@@ -257,6 +264,10 @@ public class ProxyService implements Closeable {
return proxyConfig;
}
+ public Timer getTimer() {
+ return timer;
+ }
+
public AuthenticationService getAuthenticationService() {
return authenticationService;
}
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
index c10312a..b16cc46 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
@@ -102,7 +102,7 @@ public class PulsarConnectorCache {
.setExplictLacInterval(pulsarConnectorConfig.getBookkeeperExplicitInterval())
.setStickyReadsEnabled(false)
.setReadEntryTimeout(60)
- .setThrottleValue(pulsarConnectorConfig.getBookkeeperThrottleValue())
+ .setThrottleValue(pulsarConnectorConfig.getBookkeeperThrottleValue())PulsarClientImplTest.java
.setNumIOThreads(pulsarConnectorConfig.getBookkeeperNumIOThreads())
.setNumWorkerThreads(pulsarConnectorConfig.getBookkeeperNumWorkerThreads());
[pulsar] 04/07: Fix possible name mismatch bugs when build wheel
files within docker (#10051)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a23f31ccd5f56838e512153622cebf37aa9343e6
Author: Ting Yuan <ty...@streamnative.io>
AuthorDate: Sat Mar 27 20:17:04 2021 -0500
Fix possible name mismatch bugs when build wheel files within docker (#10051)
(cherry picked from commit 510ecfa164034831ad71ade06734f5b685bfa57a)
---
pulsar-client-cpp/docker/build-wheel-file-within-docker.sh | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-client-cpp/docker/build-wheel-file-within-docker.sh b/pulsar-client-cpp/docker/build-wheel-file-within-docker.sh
index a1846d5..ade3ca0 100755
--- a/pulsar-client-cpp/docker/build-wheel-file-within-docker.sh
+++ b/pulsar-client-cpp/docker/build-wheel-file-within-docker.sh
@@ -42,4 +42,4 @@ python setup.py bdist_wheel
# Audit wheel will make sure no external dependencies are needed for
# the shared library and that only symbols supported by most linux
# distributions are used.
-auditwheel repair dist/pulsar_client-*-$PYTHON_SPEC-linux_x86_64.whl
+auditwheel repair dist/pulsar_client*-$PYTHON_SPEC-linux_x86_64.whl
[pulsar] 06/07: Fix checkstyle issue
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0eb7311169559f3aedf2a2997eb23628537325d6
Author: penghui <pe...@apache.org>
AuthorDate: Tue Mar 30 22:18:30 2021 +0800
Fix checkstyle issue
---
.../main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
index 1530c79..c10312a 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
@@ -112,7 +112,8 @@ public class PulsarConnectorCache {
pulsarConnectorConfig.getManagedLedgerNumWorkerThreads());
managedLedgerFactoryConfig.setNumManagedLedgerSchedulerThreads(
pulsarConnectorConfig.getManagedLedgerNumSchedulerThreads());
- return new ManagedLedgerFactoryImpl(bkClientConfiguration, pulsarConnectorConfig.getZookeeperUri(),managedLedgerFactoryConfig);
+ return new ManagedLedgerFactoryImpl(bkClientConfiguration, pulsarConnectorConfig.getZookeeperUri(),
+ managedLedgerFactoryConfig);
}
public ManagedLedgerConfig getManagedLedgerConfig(NamespaceName namespaceName, OffloadPolicies offloadPolicies,
[pulsar] 01/07: [Pulsar-sql]Using pulsar SQL query messages will appear `NoSuchLedger… (#9910)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit adc04f6fe481980644b26131b9ba956a769d1f3b
Author: li jinquan <40...@users.noreply.github.com>
AuthorDate: Wed Mar 24 16:55:48 2021 +0800
[Pulsar-sql]Using pulsar SQL query messages will appear `NoSuchLedger… (#9910)
* [Pulsar-sql]Using pulsar SQL query messages will appear `NoSuchLedger` when zk root directory changed (#2258)
Fixes #2258
*Motivation*
When zookeeper ledgers root path is changed, using pulsar-sql to query messages will cause `BKNoSuchLedgerExistsException`.
*Modifications*
To use new DefaultBkFactory(clientConfiguration)，so that zk will be null in Bookeeper constructor;(Bookeeper.java row 113)
when metadataDriver will be initialized(Bookeeper.java row 167),zookeeper conection is null; we can jump to another branch.If we have done the above steps,
finally, zkServers will be localhost:2181 rather than localhost:2181/pulsar in row 168(ZKMetadataDriverBase.java); the path that we use to get ledger is localhost:2181/pulsar/ledger/00/0000/L0001 rather than localhost:2181/pulsar/pulsar/ledger/00/0000/L0001;
* [Pulsar-sql]Using pulsar SQL query messages will appear `NoSuchLedger` when zk root directory changed (#2258)
Fixes #2258
*Motivation*
When zookeeper ledgers root path is changed, using pulsar-sql to query messages will cause `BKNoSuchLedgerExistsException`.
*Modifications*
To use new DefaultBkFactory(clientConfiguration)，so that zk will be null in Bookeeper constructor;(Bookeeper.java row 113)
when metadataDriver will be initialized(Bookeeper.java row 167),zookeeper conection is null; we can jump to another branch.If we have done the above steps,
finally, zkServers will be localhost:2181 rather than localhost:2181/pulsar in row 168(ZKMetadataDriverBase.java); the path that we use to get ledger is localhost:2181/pulsar/ledger/00/0000/L0001 rather than localhost:2181/pulsar/pulsar/ledger/00/0000/L0001;
* [Pulsar-sql]Using pulsar SQL query messages will appear `NoSuchLedger` when zk root directory changed (#2258)
Fixes #2258
*Motivation*
When zookeeper ledgers root path is changed, using pulsar-sql to query messages will cause `BKNoSuchLedgerExistsException`.
*Modifications*
To use new DefaultBkFactory(clientConfiguration)，so that zk will be null in Bookeeper constructor;(Bookeeper.java row 113)
when metadataDriver will be initialized(Bookeeper.java row 167),zookeeper conection is null; we can jump to another branch.If we have done the above steps,
finally, zkServers will be localhost:2181 rather than localhost:2181/pulsar in row 168(ZKMetadataDriverBase.java); the path that we use to get ledger is localhost:2181/pulsar/ledger/00/0000/L0001 rather than localhost:2181/pulsar/pulsar/ledger/00/0000/L0001;
Co-authored-by: 919364728@qq.com <1314520Ljq-->
(cherry picked from commit b1ba8e5c87e86ab00bf9b8e854cca002e7e0fe0f)
---
.../mledger/impl/ManagedLedgerFactoryImpl.java | 2 +-
.../ManagedLedgerFactoryChangeLedgerPathTest.java | 36 ++++++++++++++++++++++
.../pulsar/sql/presto/PulsarConnectorCache.java | 2 +-
3 files changed, 38 insertions(+), 2 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
index 7efb471..3354fab 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
@@ -145,7 +145,7 @@ public class ManagedLedgerFactoryImpl implements ManagedLedgerFactory {
zkc, config, NullStatsLogger.INSTANCE);
}
- private ManagedLedgerFactoryImpl(ClientConfiguration clientConfiguration, String zkConnection, ManagedLedgerFactoryConfig config) throws Exception {
+ public ManagedLedgerFactoryImpl(ClientConfiguration clientConfiguration, String zkConnection, ManagedLedgerFactoryConfig config) throws Exception {
this(new DefaultBkFactory(clientConfiguration),
true,
ZooKeeperClient.newBuilder()
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryChangeLedgerPathTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryChangeLedgerPathTest.java
index 1b596d7..c70f5bb 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryChangeLedgerPathTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryChangeLedgerPathTest.java
@@ -22,6 +22,7 @@ import java.util.List;
import org.apache.bookkeeper.common.allocator.PoolingPolicy;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.*;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
@@ -75,4 +76,39 @@ public class ManagedLedgerFactoryChangeLedgerPathTest extends BookKeeperClusterT
}
factory.shutdown();
}
+ @Test()
+ public void testChangeZKPath2() throws Exception {
+ ClientConfiguration configuration = new ClientConfiguration();
+ String zkConnectString = zkUtil.getZooKeeperConnectString() + "/test";
+ configuration.setMetadataServiceUri("zk://" + zkConnectString + "/ledgers");
+ configuration.setUseV2WireProtocol(true);
+ configuration.setEnableDigestTypeAutodetection(true);
+ configuration.setAllocatorPoolingPolicy(PoolingPolicy.UnpooledHeap);
+
+ ManagedLedgerFactoryConfig managedLedgerFactoryConfig = new ManagedLedgerFactoryConfig();
+ ManagedLedgerFactory factory = new ManagedLedgerFactoryImpl(configuration, zkConnectString,managedLedgerFactoryConfig);
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setEnsembleSize(1)
+ .setWriteQuorumSize(1)
+ .setAckQuorumSize(1)
+ .setMetadataAckQuorumSize(1)
+ .setMetadataAckQuorumSize(1);
+ ManagedLedger ledger = factory.open("test-ledger", config);
+ ManagedCursor cursor = ledger.openCursor("test-c1");
+
+ for (int i = 0; i < 10; i++) {
+ String entry = "entry" + i;
+ ledger.addEntry(entry.getBytes("UTF8"));
+ }
+
+ List<Entry> entryList = cursor.readEntries(10);
+ Assert.assertEquals(10, entryList.size());
+
+ for (int i = 0; i < 10; i++) {
+ Entry entry = entryList.get(i);
+ Assert.assertEquals(("entry" + i).getBytes("UTF8"), entry.getData());
+ }
+ factory.shutdown();
+ }
}
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
index ec0e3b3..1530c79 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
@@ -112,7 +112,7 @@ public class PulsarConnectorCache {
pulsarConnectorConfig.getManagedLedgerNumWorkerThreads());
managedLedgerFactoryConfig.setNumManagedLedgerSchedulerThreads(
pulsarConnectorConfig.getManagedLedgerNumSchedulerThreads());
- return new ManagedLedgerFactoryImpl(bkClientConfiguration, managedLedgerFactoryConfig);
+ return new ManagedLedgerFactoryImpl(bkClientConfiguration, pulsarConnectorConfig.getZookeeperUri(),managedLedgerFactoryConfig);
}
public ManagedLedgerConfig getManagedLedgerConfig(NamespaceName namespaceName, OffloadPolicies offloadPolicies,
[pulsar] 05/07: Upgrade Netty version to 4.1.60.final (#10073)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8ed886e81cd50b23c0f8801032978515c4942c7c
Author: Ali Ahmed <al...@gmail.com>
AuthorDate: Mon Mar 29 18:47:55 2021 -0700
Upgrade Netty version to 4.1.60.final (#10073)
Co-authored-by: Ali Ahmed <al...@splunk.com>
(cherry picked from commit ae606b9e7d35bb524ccdb5f6f5b0c6a9d061a939)
---
distribution/server/src/assemble/LICENSE.bin.txt | 36 ++++++++++++------------
pom.xml | 4 +--
pulsar-sql/presto-distribution/LICENSE | 28 +++++++++---------
3 files changed, 34 insertions(+), 34 deletions(-)
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index b1815e2..776f3b4 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -357,24 +357,24 @@ The Apache Software License, Version 2.0
- org.apache.commons-commons-compress-1.19.jar
- org.apache.commons-commons-lang3-3.6.jar
* Netty
- - io.netty-netty-buffer-4.1.51.Final.jar
- - io.netty-netty-codec-4.1.51.Final.jar
- - io.netty-netty-codec-dns-4.1.51.Final.jar
- - io.netty-netty-codec-http-4.1.51.Final.jar
- - io.netty-netty-codec-http2-4.1.51.Final.jar
- - io.netty-netty-codec-socks-4.1.51.Final.jar
- - io.netty-netty-codec-haproxy-4.1.51.Final.jar
- - io.netty-netty-common-4.1.51.Final.jar
- - io.netty-netty-handler-4.1.51.Final.jar
- - io.netty-netty-handler-proxy-4.1.51.Final.jar
- - io.netty-netty-resolver-4.1.51.Final.jar
- - io.netty-netty-resolver-dns-4.1.51.Final.jar
- - io.netty-netty-transport-4.1.51.Final.jar
- - io.netty-netty-transport-native-epoll-4.1.51.Final-linux-x86_64.jar
- - io.netty-netty-transport-native-epoll-4.1.51.Final.jar
- - io.netty-netty-transport-native-unix-common-4.1.51.Final.jar
- - io.netty-netty-transport-native-unix-common-4.1.51.Final-linux-x86_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.33.Final.jar
+ - io.netty-netty-buffer-4.1.60.Final.jar
+ - io.netty-netty-codec-4.1.60.Final.jar
+ - io.netty-netty-codec-dns-4.1.60.Final.jar
+ - io.netty-netty-codec-http-4.1.60.Final.jar
+ - io.netty-netty-codec-http2-4.1.60.Final.jar
+ - io.netty-netty-codec-socks-4.1.60.Final.jar
+ - io.netty-netty-codec-haproxy-4.1.60.Final.jar
+ - io.netty-netty-common-4.1.60.Final.jar
+ - io.netty-netty-handler-4.1.60.Final.jar
+ - io.netty-netty-handler-proxy-4.1.60.Final.jar
+ - io.netty-netty-resolver-4.1.60.Final.jar
+ - io.netty-netty-resolver-dns-4.1.60.Final.jar
+ - io.netty-netty-transport-4.1.60.Final.jar
+ - io.netty-netty-transport-native-epoll-4.1.60.Final-linux-x86_64.jar
+ - io.netty-netty-transport-native-epoll-4.1.60.Final.jar
+ - io.netty-netty-transport-native-unix-common-4.1.60.Final.jar
+ - io.netty-netty-transport-native-unix-common-4.1.60.Final-linux-x86_64.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.36.Final.jar
* Prometheus client
- io.prometheus-simpleclient-0.5.0.jar
- io.prometheus-simpleclient_common-0.5.0.jar
diff --git a/pom.xml b/pom.xml
index dbcc73e..356322b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -100,8 +100,8 @@ flexible messaging model and an intuitive client API.</description>
<bookkeeper.version>4.12.0</bookkeeper.version>
<zookeeper.version>3.5.7</zookeeper.version>
- <netty.version>4.1.51.Final</netty.version>
- <netty-tc-native.version>2.0.33.Final</netty-tc-native.version>
+ <netty.version>4.1.60.Final</netty.version>
+ <netty-tc-native.version>2.0.36.Final</netty-tc-native.version>
<jetty.version>9.4.35.v20201120</jetty.version>
<jersey.version>2.31</jersey.version>
<athenz.version>1.8.38</athenz.version>
diff --git a/pulsar-sql/presto-distribution/LICENSE b/pulsar-sql/presto-distribution/LICENSE
index e866d53..8f4eef1 100644
--- a/pulsar-sql/presto-distribution/LICENSE
+++ b/pulsar-sql/presto-distribution/LICENSE
@@ -233,21 +233,21 @@ The Apache Software License, Version 2.0
- commons-lang3-3.6.jar
* Netty
- netty-3.10.6.Final.jar
- - netty-buffer-4.1.51.Final.jar
- - netty-codec-4.1.51.Final.jar
- - netty-codec-dns-4.1.51.Final.jar
- - netty-codec-http-4.1.51.Final.jar
- - netty-codec-haproxy-4.1.51.Final.jar
- - netty-common-4.1.51.Final.jar
- - netty-handler-4.1.51.Final.jar
+ - netty-buffer-4.1.60.Final.jar
+ - netty-codec-4.1.60.Final.jar
+ - netty-codec-dns-4.1.60.Final.jar
+ - netty-codec-http-4.1.60.Final.jar
+ - netty-codec-haproxy-4.1.60.Final.jar
+ - netty-common-4.1.60.Final.jar
+ - netty-handler-4.1.60.Final.jar
- netty-reactive-streams-2.0.4.jar
- - netty-resolver-4.1.51.Final.jar
- - netty-resolver-dns-4.1.51.Final.jar
- - netty-tcnative-boringssl-static-2.0.33.Final.jar
- - netty-transport-4.1.51.Final.jar
- - netty-transport-native-epoll-4.1.51.Final-linux-x86_64.jar
- - netty-transport-native-unix-common-4.1.51.Final.jar
- - netty-transport-native-unix-common-4.1.51.Final-linux-x86_64.jar
+ - netty-resolver-4.1.60.Final.jar
+ - netty-resolver-dns-4.1.60.Final.jar
+ - netty-tcnative-boringssl-static-2.0.36.Final.jar
+ - netty-transport-4.1.60.Final.jar
+ - netty-transport-native-epoll-4.1.60.Final-linux-x86_64.jar
+ - netty-transport-native-unix-common-4.1.60.Final.jar
+ - netty-transport-native-unix-common-4.1.60.Final-linux-x86_64.jar
* Joda Time
- joda-time-2.10.5.jar
- joda-time-2.10.1.jar
[pulsar] 03/07: bugfix for wrong timeunit in updating
lastLedgerCreationInitiationTimestamp (#10049)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5d00904c56f78c97460720e5b984a893231fd312
Author: Zhanpeng Wu <zh...@qq.com>
AuthorDate: Sat Mar 27 07:30:06 2021 +0800
bugfix for wrong timeunit in updating lastLedgerCreationInitiationTimestamp (#10049)
Co-authored-by: wuzhanpeng <wu...@bigo.sg>
(cherry picked from commit 1b8da9245e111718acb337bce310ffdef7b30496)
---
.../main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 754815c..66d4c71 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -1472,7 +1472,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
synchronized void createLedgerAfterClosed() {
STATE_UPDATER.set(this, State.CreatingLedger);
- this.lastLedgerCreationInitiationTimestamp = System.nanoTime();
+ this.lastLedgerCreationInitiationTimestamp = System.currentTimeMillis();
mbean.startDataLedgerCreateOp();
asyncCreateLedger(bookKeeper, config, digestType, this, Collections.emptyMap());
}