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 2022/01/27 07:19:40 UTC
[pulsar] branch branch-2.8 updated (ed771af -> 5d796e3)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from ed771af [Cherry-pick] Pass client builder to debezium database history (#12112)
new b85d06e Upgrade BK to 4.14.4 and Grpc to 1.42.1 (#13714)
new 5a491c4 Issue 12019 Fix flaky test ManagedLedgerTest.testMaximumRolloverTime (#12030)
new a669b25 Getting rid of CVes brought in with aerospike (#13819)
new 444d2a3 Getting rid of CVEs in batch-data-generator (#13820)
new f2bfe92 Getitng rid of CVEs in InfluxDB connector (#13821)
new 7751597 Getting rid of CVEs in Solr connector (#13822)
new 76fac61 NPE when get isAllowAutoUploadSchema (#13831)
new ee4ac7f6 [C++] Fix hasMessageAvailable returns wrong value for last message (#13883)
new d4bf415 Fix batch message ack does not decrease the unacked-msg count. (#13383)
new 9601de1 [Broker] Fix read schema compatibility strategy priority (#13938)
new 5d796e3 Fix cherry-pick issue
The 11 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:
conf/broker.conf | 8 +-
distribution/server/src/assemble/LICENSE.bin.txt | 94 +++++++--------
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 7 +-
pom.xml | 10 +-
.../apache/pulsar/broker/ServiceConfiguration.java | 11 +-
.../apache/pulsar/broker/admin/AdminResource.java | 15 +++
.../pulsar/broker/admin/impl/NamespacesBase.java | 14 +--
.../broker/admin/impl/SchemasResourceBase.java | 42 ++-----
.../pulsar/broker/service/AbstractTopic.java | 18 +--
.../org/apache/pulsar/broker/service/Consumer.java | 129 ++++++++++++++++++---
.../org/apache/pulsar/broker/service/Topic.java | 6 +
.../broker/admin/AdminApiSchemaAutoUpdateTest.java | 22 ++--
.../pulsar/broker/admin/AdminApiSchemaTest.java | 53 +++++++++
.../pulsar/broker/service/BatchMessageTest.java | 2 +-
.../BatchMessageWithBatchIndexLevelTest.java | 108 +++++++++++++++++
.../PersistentDispatcherFailoverConsumerTest.java | 79 ++++++-------
.../SchemaCompatibilityCheckTest.java | 6 +-
.../SchemaTypeCompatibilityCheckTest.java | 8 +-
.../pulsar/common/policies/data/Policies.java | 3 +-
.../policies/data/SchemaCompatibilityStrategy.java | 5 +-
pulsar-client-cpp/lib/ClientConnection.cc | 23 ++--
pulsar-client-cpp/lib/ClientConnection.h | 5 +-
pulsar-client-cpp/lib/ConsumerImpl.cc | 119 ++++++++++++-------
pulsar-client-cpp/lib/ConsumerImpl.h | 23 ++--
pulsar-client-cpp/lib/GetLastMessageIdResponse.h | 56 +++++++++
...kGroupingTrackerDisabled.cc => MessageIdUtil.h} | 31 ++---
pulsar-client-cpp/lib/ReaderImpl.cc | 6 +
pulsar-client-cpp/tests/MessageIdTest.cc | 22 ++++
pulsar-client-cpp/tests/ReaderTest.cc | 45 +++++++
pulsar-io/batch-data-generator/pom.xml | 6 +
pulsar-io/influxdb/pom.xml | 4 +-
pulsar-io/solr/pom.xml | 2 +-
pulsar-sql/presto-distribution/LICENSE | 24 ++--
33 files changed, 717 insertions(+), 289 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
create mode 100644 pulsar-client-cpp/lib/GetLastMessageIdResponse.h
copy pulsar-client-cpp/lib/{AckGroupingTrackerDisabled.cc => MessageIdUtil.h} (57%)
[pulsar] 11/11: Fix cherry-pick 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.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5d796e3515b51a9e9fea3dde785f2b94484b7a64
Author: penghui <pe...@apache.org>
AuthorDate: Thu Jan 27 14:31:56 2022 +0800
Fix cherry-pick issue
---
.../src/main/java/org/apache/pulsar/broker/service/Topic.java | 6 ++++++
1 file changed, 6 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
index e9638f3..a7280ca 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java
@@ -274,4 +274,10 @@ public interface Topic {
*/
CompletableFuture<Void> truncate();
+ /**
+ * Get BrokerService.
+ * @return
+ */
+ BrokerService getBrokerService();
+
}
[pulsar] 05/11: Getitng rid of CVEs in InfluxDB connector (#13821)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f2bfe9213af5879f423eb781e9f8307304222090
Author: Andrey Yegorov <86...@users.noreply.github.com>
AuthorDate: Wed Jan 19 23:52:23 2022 -0800
Getitng rid of CVEs in InfluxDB connector (#13821)
CVE-2019-20933
CVE-2018-1000850
(cherry picked from commit 85746390c4421cce710ec6988d1747fd5a3f7ade)
---
pulsar-io/influxdb/pom.xml | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/pulsar-io/influxdb/pom.xml b/pulsar-io/influxdb/pom.xml
index d0a3858..0beccde 100644
--- a/pulsar-io/influxdb/pom.xml
+++ b/pulsar-io/influxdb/pom.xml
@@ -51,13 +51,13 @@
<dependency>
<groupId>com.influxdb</groupId>
<artifactId>influxdb-client-java</artifactId>
- <version>1.6.0</version>
+ <version>4.0.0</version>
</dependency>
<dependency>
<groupId>org.influxdb</groupId>
<artifactId>influxdb-java</artifactId>
- <version>2.7</version>
+ <version>2.22</version>
<exclusions>
<exclusion>
<groupId>com.squareup.okhttp3</groupId>
[pulsar] 08/11: [C++] Fix hasMessageAvailable returns wrong value for last message (#13883)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ee4ac7f6659f86091cf657199e5d0d38a26bc5c8
Author: Yunze Xu <xy...@163.com>
AuthorDate: Mon Jan 24 12:06:00 2022 +0800
[C++] Fix hasMessageAvailable returns wrong value for last message (#13883)
In C++ client, there is a corner case that when a reader's start message ID is the last message of a topic, `hasMessageAvailable` returns true. However, it should return false because the start message ID is exclusive and in this case `readNext` would never return a message unless new messages arrived.
The current C++ implementation of `hasMessageAvailable` is from long days ago and has many problems. So this PR migrates the Java implementation of `hasMessageAvailable` to C++ client.
Since after the modifications we need to access `startMessageId` in `hasMessageAvailable`, which is called in a different thread from `connectionOpened` that might modify `startMessageId`. We use a common mutex `mutexForMessageIds` to protect the access to `lastDequedMessageId_` and `lastMessageIdInBroker_`.
To fix the original tests when `startMessageId` is latest, this PR adds a `GetLastMessageIdResponse` as the response of `GetLastMessageId` request. The `GetLastMessageIdResponse` contains the `consumer_mark_delete_position` introduced from https://github.com/apache/pulsar/pull/9652 to compare with `last_message_id` when `startMessageId` is latest.
This change added tests `ReaderTest#testHasMessageAvailableWhenCreated` and `MessageIdTest# testCompareLedgerAndEntryId`.
(cherry picked from commit e50493ea17dd5f2f9d4527d74cc4f40e12439df2)
---
pulsar-client-cpp/lib/ClientConnection.cc | 23 ++--
pulsar-client-cpp/lib/ClientConnection.h | 5 +-
pulsar-client-cpp/lib/ConsumerImpl.cc | 119 +++++++++++++--------
pulsar-client-cpp/lib/ConsumerImpl.h | 23 ++--
pulsar-client-cpp/lib/GetLastMessageIdResponse.h | 56 ++++++++++
.../MessageIdTest.cc => lib/MessageIdUtil.h} | 33 +++---
pulsar-client-cpp/lib/ReaderImpl.cc | 6 ++
pulsar-client-cpp/tests/MessageIdTest.cc | 22 ++++
pulsar-client-cpp/tests/ReaderTest.cc | 45 ++++++++
9 files changed, 249 insertions(+), 83 deletions(-)
diff --git a/pulsar-client-cpp/lib/ClientConnection.cc b/pulsar-client-cpp/lib/ClientConnection.cc
index 21fdd64..9f4ae6d 100644
--- a/pulsar-client-cpp/lib/ClientConnection.cc
+++ b/pulsar-client-cpp/lib/ClientConnection.cc
@@ -37,6 +37,7 @@
#include "ProducerImpl.h"
#include "ConsumerImpl.h"
#include "checksum/ChecksumProvider.h"
+#include "MessageIdUtil.h"
DECLARE_LOG_OBJECT()
@@ -1052,7 +1053,7 @@ void ClientConnection::handleIncomingCommand() {
PendingGetLastMessageIdRequestsMap::iterator it =
pendingGetLastMessageIdRequests_.find(error.request_id());
if (it != pendingGetLastMessageIdRequests_.end()) {
- Promise<Result, MessageId> getLastMessageIdPromise = it->second;
+ auto getLastMessageIdPromise = it->second;
pendingGetLastMessageIdRequests_.erase(it);
lock.unlock();
@@ -1168,15 +1169,18 @@ void ClientConnection::handleIncomingCommand() {
pendingGetLastMessageIdRequests_.find(getLastMessageIdResponse.request_id());
if (it != pendingGetLastMessageIdRequests_.end()) {
- Promise<Result, MessageId> getLastMessageIdPromise = it->second;
+ auto getLastMessageIdPromise = it->second;
pendingGetLastMessageIdRequests_.erase(it);
lock.unlock();
- MessageIdData messageIdData = getLastMessageIdResponse.last_message_id();
- MessageId messageId = MessageId(messageIdData.partition(), messageIdData.ledgerid(),
- messageIdData.entryid(), messageIdData.batch_index());
-
- getLastMessageIdPromise.setValue(messageId);
+ if (getLastMessageIdResponse.has_consumer_mark_delete_position()) {
+ getLastMessageIdPromise.setValue(
+ {toMessageId(getLastMessageIdResponse.last_message_id()),
+ toMessageId(getLastMessageIdResponse.consumer_mark_delete_position())});
+ } else {
+ getLastMessageIdPromise.setValue(
+ {toMessageId(getLastMessageIdResponse.last_message_id())});
+ }
} else {
lock.unlock();
LOG_WARN(
@@ -1585,9 +1589,10 @@ Commands::ChecksumType ClientConnection::getChecksumType() const {
return getServerProtocolVersion() >= proto::v6 ? Commands::Crc32c : Commands::None;
}
-Future<Result, MessageId> ClientConnection::newGetLastMessageId(uint64_t consumerId, uint64_t requestId) {
+Future<Result, GetLastMessageIdResponse> ClientConnection::newGetLastMessageId(uint64_t consumerId,
+ uint64_t requestId) {
Lock lock(mutex_);
- Promise<Result, MessageId> promise;
+ Promise<Result, GetLastMessageIdResponse> promise;
if (isClosed()) {
lock.unlock();
LOG_ERROR(cnxString_ << " Client is not connected to the broker");
diff --git a/pulsar-client-cpp/lib/ClientConnection.h b/pulsar-client-cpp/lib/ClientConnection.h
index 4ba99fa..150a37b 100644
--- a/pulsar-client-cpp/lib/ClientConnection.h
+++ b/pulsar-client-cpp/lib/ClientConnection.h
@@ -46,6 +46,7 @@
#include <set>
#include <lib/BrokerConsumerStatsImpl.h>
#include "lib/PeriodicTask.h"
+#include "lib/GetLastMessageIdResponse.h"
using namespace pulsar;
@@ -156,7 +157,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
Future<Result, BrokerConsumerStatsImpl> newConsumerStats(uint64_t consumerId, uint64_t requestId);
- Future<Result, MessageId> newGetLastMessageId(uint64_t consumerId, uint64_t requestId);
+ Future<Result, GetLastMessageIdResponse> newGetLastMessageId(uint64_t consumerId, uint64_t requestId);
Future<Result, NamespaceTopicsPtr> newGetTopicsOfNamespace(const std::string& nsName, uint64_t requestId);
@@ -305,7 +306,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
typedef std::map<uint64_t, Promise<Result, BrokerConsumerStatsImpl>> PendingConsumerStatsMap;
PendingConsumerStatsMap pendingConsumerStatsMap_;
- typedef std::map<long, Promise<Result, MessageId>> PendingGetLastMessageIdRequestsMap;
+ typedef std::map<long, Promise<Result, GetLastMessageIdResponse>> PendingGetLastMessageIdRequestsMap;
PendingGetLastMessageIdRequestsMap pendingGetLastMessageIdRequests_;
typedef std::map<long, Promise<Result, NamespaceTopicsPtr>> PendingGetNamespaceTopicsMap;
diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc
index 4b6479d..3605d4e 100644
--- a/pulsar-client-cpp/lib/ConsumerImpl.cc
+++ b/pulsar-client-cpp/lib/ConsumerImpl.cc
@@ -25,6 +25,7 @@
#include "pulsar/Result.h"
#include "pulsar/MessageId.h"
#include "Utils.h"
+#include "MessageIdUtil.h"
#include "AckGroupingTracker.h"
#include "AckGroupingTrackerEnabled.h"
#include "AckGroupingTrackerDisabled.h"
@@ -50,7 +51,6 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic,
hasParent_(hasParent),
consumerTopicType_(consumerTopicType),
subscriptionMode_(subscriptionMode),
- startMessageId_(startMessageId),
// This is the initial capacity of the queue
incomingMessages_(std::max(config_.getReceiverQueueSize(), 1)),
availablePermits_(0),
@@ -62,7 +62,7 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic,
negativeAcksTracker_(client, *this, conf),
ackGroupingTrackerPtr_(std::make_shared<AckGroupingTracker>()),
readCompacted_(conf.isReadCompacted()),
- lastMessageInBroker_(Optional<MessageId>::of(MessageId())) {
+ startMessageId_(startMessageId),
std::stringstream consumerStrStream;
consumerStrStream << "[" << topic_ << ", " << subscription_ << ", " << consumerId_ << "] ";
consumerStr_ = consumerStrStream.str();
@@ -158,29 +158,38 @@ void ConsumerImpl::start() {
void ConsumerImpl::connectionOpened(const ClientConnectionPtr& cnx) {
Lock lock(mutex_);
- if (state_ == Closed) {
- lock.unlock();
+ const auto state = state_;
+ lock.unlock();
+ if (state == Closed) {
LOG_DEBUG(getName() << "connectionOpened : Consumer is already closed");
return;
}
- Optional<MessageId> firstMessageInQueue = clearReceiveQueue();
- unAckedMessageTrackerPtr_->clear();
- batchAcknowledgementTracker_.clear();
+<<<<<<< HEAD
+=======
+ // Register consumer so that we can handle other incomming commands (e.g. ACTIVE_CONSUMER_CHANGE) after
+ // sending the subscribe request.
+ cnx->registerConsumer(consumerId_, shared_from_this());
+ Lock lockForMessageId(mutexForMessageId_);
+>>>>>>> e50493ea17d ([C++] Fix hasMessageAvailable returns wrong value for last message (#13883))
+ Optional<MessageId> firstMessageInQueue = clearReceiveQueue();
if (subscriptionMode_ == Commands::SubscriptionModeNonDurable) {
// Update startMessageId so that we can discard messages after delivery
// restarts
startMessageId_ = firstMessageInQueue;
}
+ const auto startMessageId = startMessageId_;
+ lockForMessageId.unlock();
- lock.unlock();
+ unAckedMessageTrackerPtr_->clear();
+ batchAcknowledgementTracker_.clear();
ClientImplPtr client = client_.lock();
uint64_t requestId = client->newRequestId();
SharedBuffer cmd = Commands::newSubscribe(
topic_, subscription_, consumerId_, requestId, getSubType(), consumerName_, subscriptionMode_,
- startMessageId_, readCompacted_, config_.getProperties(), config_.getSchema(), getInitialPosition(),
+ startMessageId, readCompacted_, config_.getProperties(), config_.getSchema(), getInitialPosition(),
config_.isReplicateSubscriptionStateEnabled(), config_.getKeySharedPolicy());
cnx->sendRequestWithId(cmd, requestId)
.addListener(
@@ -416,6 +425,9 @@ uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(const ClientConnection
batchAcknowledgementTracker_.receivedMessage(batchedMessage);
LOG_DEBUG("Received Batch messages of size - " << batchSize
<< " -- msgId: " << batchedMessage.getMessageId());
+ Lock lock(mutexForMessageId_);
+ const auto startMessageId = startMessageId_;
+ lock.unlock();
int skippedMessages = 0;
@@ -425,14 +437,14 @@ uint32_t ConsumerImpl::receiveIndividualMessagesFromBatch(const ClientConnection
msg.impl_->setRedeliveryCount(redeliveryCount);
msg.impl_->setTopicName(batchedMessage.getTopicName());
- if (startMessageId_.is_present()) {
+ if (startMessageId.is_present()) {
const MessageId& msgId = msg.getMessageId();
// If we are receiving a batch message, we need to discard messages that were prior
// to the startMessageId
- if (msgId.ledgerId() == startMessageId_.value().ledgerId() &&
- msgId.entryId() == startMessageId_.value().entryId() &&
- msgId.batchIndex() <= startMessageId_.value().batchIndex()) {
+ if (msgId.ledgerId() == startMessageId.value().ledgerId() &&
+ msgId.entryId() == startMessageId.value().entryId() &&
+ msgId.batchIndex() <= startMessageId.value().batchIndex()) {
LOG_DEBUG(getName() << "Ignoring message from before the startMessageId"
<< msg.getMessageId());
++skippedMessages;
@@ -563,7 +575,7 @@ void ConsumerImpl::internalListener() {
trackMessage(msg);
try {
consumerStatsBasePtr_->receivedMessage(msg, ResultOk);
- lastDequedMessage_ = Optional<MessageId>::of(msg.getMessageId());
+ lastDequedMessageId_ = msg.getMessageId();
messageListener_(Consumer(shared_from_this()), msg);
} catch (const std::exception& e) {
LOG_ERROR(getName() << "Exception thrown from listener" << e.what());
@@ -697,8 +709,9 @@ Result ConsumerImpl::receiveHelper(Message& msg, int timeout) {
}
void ConsumerImpl::messageProcessed(Message& msg, bool track) {
- Lock lock(mutex_);
- lastDequedMessage_ = Optional<MessageId>::of(msg.getMessageId());
+ Lock lock(mutexForMessageId_);
+ lastDequedMessageId_ = msg.getMessageId();
+ lock.unlock();
ClientConnectionPtr currentCnx = getCnx().lock();
if (currentCnx && msg.impl_->cnx_ != currentCnx.get()) {
@@ -730,11 +743,11 @@ Optional<MessageId> ConsumerImpl::clearReceiveQueue() {
previousMessageId = MessageId(-1, nextMessageId.ledgerId(), nextMessageId.entryId() - 1, -1);
}
return Optional<MessageId>::of(previousMessageId);
- } else if (lastDequedMessage_.is_present()) {
+ } else if (lastDequedMessageId_ != MessageId::earliest()) {
// If the queue was empty we need to restart from the message just after the last one that has been
// dequeued
// in the past
- return lastDequedMessage_;
+ return Optional<MessageId>::of(lastDequedMessageId_);
} else {
// No message was received or dequeued by this consumer. Next message would still be the
// startMessageId
@@ -1070,6 +1083,9 @@ void ConsumerImpl::brokerConsumerStatsListener(Result res, BrokerConsumerStatsIm
void ConsumerImpl::handleSeek(Result result, ResultCallback callback) {
if (result == ResultOk) {
+ Lock lock(mutexForMessageId_);
+ lastDequedMessageId_ = MessageId::earliest();
+ lock.unlock();
LOG_INFO(getName() << "Seek successfully");
} else {
LOG_ERROR(getName() << "Failed to seek: " << strResult(result));
@@ -1144,37 +1160,42 @@ void ConsumerImpl::seekAsync(uint64_t timestamp, ResultCallback callback) {
bool ConsumerImpl::isReadCompacted() { return readCompacted_; }
+inline bool hasMoreMessages(const MessageId& lastMessageIdInBroker, const MessageId& messageId) {
+ return lastMessageIdInBroker > messageId && lastMessageIdInBroker.entryId() != -1;
+}
+
void ConsumerImpl::hasMessageAvailableAsync(HasMessageAvailableCallback callback) {
- MessageId lastDequed = this->lastMessageIdDequed();
- MessageId lastInBroker = this->lastMessageIdInBroker();
- if (lastInBroker > lastDequed && lastInBroker.entryId() != -1) {
- callback(ResultOk, true);
- return;
- }
+ Lock lock(mutexForMessageId_);
+ const auto messageId =
+ (lastDequedMessageId_ == MessageId::earliest()) ? startMessageId_.value() : lastDequedMessageId_;
- getLastMessageIdAsync([lastDequed, callback](Result result, MessageId messageId) {
- if (result == ResultOk) {
- if (messageId > lastDequed && messageId.entryId() != -1) {
- callback(ResultOk, true);
+ if (messageId == MessageId::latest()) {
+ lock.unlock();
+ getLastMessageIdAsync([callback](Result result, const GetLastMessageIdResponse& response) {
+ if (result != ResultOk) {
+ callback(result, {});
+ return;
+ }
+ if (response.hasMarkDeletePosition() && response.getLastMessageId().entryId() >= 0) {
+ // We only care about comparing ledger ids and entry ids as mark delete position doesn't have
+ // other ids such as batch index
+ callback(ResultOk, compareLedgerAndEntryId(response.getMarkDeletePosition(),
+ response.getLastMessageId()) < 0);
} else {
callback(ResultOk, false);
}
- } else {
- callback(result, false);
- }
- });
-}
-
-void ConsumerImpl::brokerGetLastMessageIdListener(Result res, MessageId messageId,
- BrokerGetLastMessageIdCallback callback) {
- Lock lock(mutex_);
- if (messageId > lastMessageIdInBroker()) {
- lastMessageInBroker_ = Optional<MessageId>::of(messageId);
- lock.unlock();
- callback(res, messageId);
+ });
} else {
+ if (hasMoreMessages(lastMessageIdInBroker_, messageId)) {
+ lock.unlock();
+ callback(ResultOk, true);
+ return;
+ }
lock.unlock();
- callback(res, lastMessageIdInBroker());
+
+ getLastMessageIdAsync([callback, messageId](Result result, const GetLastMessageIdResponse& response) {
+ callback(result, (result == ResultOk) && hasMoreMessages(response.getLastMessageId(), messageId));
+ });
}
}
@@ -1198,9 +1219,19 @@ void ConsumerImpl::getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback
LOG_DEBUG(getName() << " Sending getLastMessageId Command for Consumer - " << getConsumerId()
<< ", requestId - " << requestId);
+ auto self = shared_from_this();
cnx->newGetLastMessageId(consumerId_, requestId)
- .addListener(std::bind(&ConsumerImpl::brokerGetLastMessageIdListener, shared_from_this(),
- std::placeholders::_1, std::placeholders::_2, callback));
+ .addListener([this, self, callback](Result result, const GetLastMessageIdResponse& response) {
+ if (result == ResultOk) {
+ LOG_DEBUG(getName() << "getLastMessageId: " << response);
+ Lock lock(mutexForMessageId_);
+ lastMessageIdInBroker_ = response.getLastMessageId();
+ lock.unlock();
+ } else {
+ LOG_ERROR(getName() << "Failed to getLastMessageId: " << result);
+ }
+ callback(result, response);
+ });
} else {
LOG_ERROR(getName() << " Operation not supported since server protobuf version "
<< cnx->getServerProtocolVersion() << " is older than proto::v12");
diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h
index b176209..2ab716f 100644
--- a/pulsar-client-cpp/lib/ConsumerImpl.h
+++ b/pulsar-client-cpp/lib/ConsumerImpl.h
@@ -33,6 +33,7 @@
#include "lib/UnAckedMessageTrackerDisabled.h"
#include "MessageCrypto.h"
#include "AckGroupingTracker.h"
+#include "GetLastMessageIdResponse.h"
#include "CompressionCodec.h"
#include <boost/dynamic_bitset.hpp>
@@ -53,7 +54,7 @@ class ExecutorService;
class ConsumerImpl;
class BatchAcknowledgementTracker;
typedef std::shared_ptr<MessageCrypto> MessageCryptoPtr;
-typedef std::function<void(Result result, MessageId messageId)> BrokerGetLastMessageIdCallback;
+typedef std::function<void(Result, const GetLastMessageIdResponse&)> BrokerGetLastMessageIdCallback;
enum ConsumerTopicType
{
@@ -186,10 +187,8 @@ class ConsumerImpl : public ConsumerImplBase,
bool hasParent_;
ConsumerTopicType consumerTopicType_;
- Commands::SubscriptionMode subscriptionMode_;
- Optional<MessageId> startMessageId_;
+ const Commands::SubscriptionMode subscriptionMode_;
- Optional<MessageId> lastDequedMessage_;
UnboundedBlockingQueue<Message> incomingMessages_;
std::queue<ReceiveCallback> pendingReceives_;
std::atomic_int availablePermits_;
@@ -210,17 +209,11 @@ class ConsumerImpl : public ConsumerImplBase,
MessageCryptoPtr msgCrypto_;
const bool readCompacted_;
- Optional<MessageId> lastMessageInBroker_;
- void brokerGetLastMessageIdListener(Result res, MessageId messageId,
- BrokerGetLastMessageIdCallback callback);
-
- const MessageId& lastMessageIdDequed() {
- return lastDequedMessage_.is_present() ? lastDequedMessage_.value() : MessageId::earliest();
- }
-
- const MessageId& lastMessageIdInBroker() {
- return lastMessageInBroker_.is_present() ? lastMessageInBroker_.value() : MessageId::earliest();
- }
+ // Make the access to `startMessageId_`, `lastDequedMessageId_` and `lastMessageIdInBroker_` thread safe
+ mutable std::mutex mutexForMessageId_;
+ Optional<MessageId> startMessageId_;
+ MessageId lastDequedMessageId_{MessageId::earliest()};
+ MessageId lastMessageIdInBroker_{MessageId::earliest()};
friend class PulsarFriend;
diff --git a/pulsar-client-cpp/lib/GetLastMessageIdResponse.h b/pulsar-client-cpp/lib/GetLastMessageIdResponse.h
new file mode 100644
index 0000000..0acb783
--- /dev/null
+++ b/pulsar-client-cpp/lib/GetLastMessageIdResponse.h
@@ -0,0 +1,56 @@
+/**
+ * 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.
+ */
+#pragma once
+
+#include <pulsar/MessageId.h>
+#include <iostream>
+
+namespace pulsar {
+
+class GetLastMessageIdResponse {
+ friend std::ostream& operator<<(std::ostream& os, const GetLastMessageIdResponse& response) {
+ os << "lastMessageId: " << response.lastMessageId_;
+ if (response.hasMarkDeletePosition_) {
+ os << ", markDeletePosition: " << response.markDeletePosition_;
+ }
+ return os;
+ }
+
+ public:
+ GetLastMessageIdResponse() = default;
+
+ GetLastMessageIdResponse(const MessageId& lastMessageId)
+ : lastMessageId_(lastMessageId), hasMarkDeletePosition_{false} {}
+
+ GetLastMessageIdResponse(const MessageId& lastMessageId, const MessageId& markDeletePosition)
+ : lastMessageId_(lastMessageId),
+ markDeletePosition_(markDeletePosition),
+ hasMarkDeletePosition_(true) {}
+
+ const MessageId& getLastMessageId() const noexcept { return lastMessageId_; }
+ const MessageId& getMarkDeletePosition() const noexcept { return markDeletePosition_; }
+ bool hasMarkDeletePosition() const noexcept { return hasMarkDeletePosition_; }
+
+ private:
+ MessageId lastMessageId_;
+ MessageId markDeletePosition_;
+ bool hasMarkDeletePosition_;
+};
+
+} // namespace pulsar
diff --git a/pulsar-client-cpp/tests/MessageIdTest.cc b/pulsar-client-cpp/lib/MessageIdUtil.h
similarity index 52%
copy from pulsar-client-cpp/tests/MessageIdTest.cc
copy to pulsar-client-cpp/lib/MessageIdUtil.h
index 06c2528..d6f80a1 100644
--- a/pulsar-client-cpp/tests/MessageIdTest.cc
+++ b/pulsar-client-cpp/lib/MessageIdUtil.h
@@ -17,21 +17,28 @@
* under the License.
*/
#include <pulsar/MessageId.h>
-#include "PulsarFriend.h"
+#include "PulsarApi.pb.h"
-#include <gtest/gtest.h>
+namespace pulsar {
-#include <string>
-
-using namespace pulsar;
-
-TEST(MessageIdTest, testSerialization) {
- MessageId msgId = PulsarFriend::getMessageId(-1, 1, 2, 3);
-
- std::string serialized;
- msgId.serialize(serialized);
+inline MessageId toMessageId(const proto::MessageIdData& messageIdData) {
+ return MessageId{messageIdData.partition(), static_cast<int64_t>(messageIdData.ledgerid()),
+ static_cast<int64_t>(messageIdData.entryid()), messageIdData.batch_index()};
+}
- MessageId deserialized = MessageId::deserialize(serialized);
+namespace internal {
+template <typename T>
+static int compare(T lhs, T rhs) {
+ return (lhs < rhs) ? -1 : ((lhs == rhs) ? 0 : 1);
+}
+} // namespace internal
- ASSERT_EQ(msgId, deserialized);
+inline int compareLedgerAndEntryId(const MessageId& lhs, const MessageId& rhs) {
+ auto result = internal::compare(lhs.ledgerId(), rhs.ledgerId());
+ if (result != 0) {
+ return result;
+ }
+ return internal::compare(lhs.entryId(), rhs.entryId());
}
+
+} // namespace pulsar
diff --git a/pulsar-client-cpp/lib/ReaderImpl.cc b/pulsar-client-cpp/lib/ReaderImpl.cc
index ccde6e7..0a7b321 100644
--- a/pulsar-client-cpp/lib/ReaderImpl.cc
+++ b/pulsar-client-cpp/lib/ReaderImpl.cc
@@ -138,6 +138,12 @@ void ReaderImpl::seekAsync(uint64_t timestamp, ResultCallback callback) {
consumer_->seekAsync(timestamp, callback);
}
+void ReaderImpl::getLastMessageIdAsync(GetLastMessageIdCallback callback) {
+ consumer_->getLastMessageIdAsync([callback](Result result, const GetLastMessageIdResponse& response) {
+ callback(result, response.getLastMessageId());
+ });
+}
+
ReaderImplWeakPtr ReaderImpl::getReaderImplWeakPtr() { return readerImplWeakPtr_; }
bool ReaderImpl::isConnected() const { return consumer_->isConnected(); }
diff --git a/pulsar-client-cpp/tests/MessageIdTest.cc b/pulsar-client-cpp/tests/MessageIdTest.cc
index 06c2528..55fa181 100644
--- a/pulsar-client-cpp/tests/MessageIdTest.cc
+++ b/pulsar-client-cpp/tests/MessageIdTest.cc
@@ -17,6 +17,7 @@
* under the License.
*/
#include <pulsar/MessageId.h>
+#include "lib/MessageIdUtil.h"
#include "PulsarFriend.h"
#include <gtest/gtest.h>
@@ -35,3 +36,24 @@ TEST(MessageIdTest, testSerialization) {
ASSERT_EQ(msgId, deserialized);
}
+
+TEST(MessageIdTest, testCompareLedgerAndEntryId) {
+ MessageId id1(-1, 2L, 1L, 0);
+ MessageId id2(-1, 2L, 1L, 1);
+ MessageId id3(-1, 2L, 2L, 0);
+ MessageId id4(-1, 3L, 0L, 0);
+ ASSERT_EQ(compareLedgerAndEntryId(id1, id2), 0);
+ ASSERT_EQ(compareLedgerAndEntryId(id1, id2), 0);
+
+ ASSERT_EQ(compareLedgerAndEntryId(id1, id3), -1);
+ ASSERT_EQ(compareLedgerAndEntryId(id3, id1), 1);
+
+ ASSERT_EQ(compareLedgerAndEntryId(id1, id4), -1);
+ ASSERT_EQ(compareLedgerAndEntryId(id4, id1), 1);
+
+ ASSERT_EQ(compareLedgerAndEntryId(id2, id4), -1);
+ ASSERT_EQ(compareLedgerAndEntryId(id4, id2), 1);
+
+ ASSERT_EQ(compareLedgerAndEntryId(id3, id4), -1);
+ ASSERT_EQ(compareLedgerAndEntryId(id4, id3), 1);
+}
diff --git a/pulsar-client-cpp/tests/ReaderTest.cc b/pulsar-client-cpp/tests/ReaderTest.cc
index d95038b..8cd535c 100644
--- a/pulsar-client-cpp/tests/ReaderTest.cc
+++ b/pulsar-client-cpp/tests/ReaderTest.cc
@@ -26,6 +26,7 @@
#include <time.h>
#include <string>
+#include <lib/Latch.h>
#include <lib/LogUtils.h>
DECLARE_LOG_OBJECT()
@@ -577,3 +578,47 @@ TEST(ReaderTest, testIsConnected) {
ASSERT_EQ(ResultOk, reader.close());
ASSERT_FALSE(reader.isConnected());
}
+
+TEST(ReaderTest, testHasMessageAvailableWhenCreated) {
+ const std::string topic = "testHasMessageAvailableWhenCreated-" + std::to_string(time(nullptr));
+ Client client(serviceUrl);
+
+ ProducerConfiguration producerConf;
+ producerConf.setBatchingMaxMessages(3);
+ Producer producer;
+ ASSERT_EQ(ResultOk, client.createProducer(topic, producerConf, producer));
+
+ std::vector<MessageId> messageIds;
+ constexpr int numMessages = 7;
+ Latch latch(numMessages);
+ for (int i = 0; i < numMessages; i++) {
+ producer.sendAsync(MessageBuilder().setContent("msg-" + std::to_string(i)).build(),
+ [i, &messageIds, &latch](Result result, const MessageId& messageId) {
+ if (result == ResultOk) {
+ LOG_INFO("Send " << i << " to " << messageId);
+ messageIds.emplace_back(messageId);
+ } else {
+ LOG_ERROR("Failed to send " << i << ": " << messageId);
+ }
+ latch.countdown();
+ });
+ }
+ latch.wait(std::chrono::seconds(3));
+ ASSERT_EQ(messageIds.size(), numMessages);
+
+ Reader reader;
+ bool hasMessageAvailable;
+
+ for (size_t i = 0; i < messageIds.size() - 1; i++) {
+ ASSERT_EQ(ResultOk, client.createReader(topic, messageIds[i], {}, reader));
+ ASSERT_EQ(ResultOk, reader.hasMessageAvailable(hasMessageAvailable));
+ EXPECT_TRUE(hasMessageAvailable);
+ }
+
+ // The start message ID is exclusive by default, so when we start at the last message, there should be no
+ // message available.
+ ASSERT_EQ(ResultOk, client.createReader(topic, messageIds.back(), {}, reader));
+ ASSERT_EQ(ResultOk, reader.hasMessageAvailable(hasMessageAvailable));
+ EXPECT_FALSE(hasMessageAvailable);
+ client.close();
+}
[pulsar] 03/11: Getting rid of CVes brought in with aerospike (#13819)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a669b258c1978beb1aa47375c50054ff6115ac9f
Author: Andrey Yegorov <86...@users.noreply.github.com>
AuthorDate: Wed Jan 19 23:51:31 2022 -0800
Getting rid of CVes brought in with aerospike (#13819)
CVE-2020-26939
CVE-2015-0886
(cherry picked from commit 8a452b733c91d6a60ddf32b403c9169be5ba7978)
---
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pom.xml b/pom.xml
index 6bf2ce9..08b40df 100644
--- a/pom.xml
+++ b/pom.xml
@@ -138,7 +138,7 @@ flexible messaging model and an intuitive client API.</description>
<sketches.version>0.8.3</sketches.version>
<hbc-core.version>2.2.0</hbc-core.version>
<cassandra-driver-core.version>3.6.0</cassandra-driver-core.version>
- <aerospike-client.version>4.4.8</aerospike-client.version>
+ <aerospike-client.version>4.4.20</aerospike-client.version>
<kafka-client.version>2.3.0</kafka-client.version>
<rabbitmq-client.version>5.1.1</rabbitmq-client.version>
<aws-sdk.version>1.11.774</aws-sdk.version>
[pulsar] 09/11: Fix batch message ack does not decrease the unacked-msg count. (#13383)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d4bf4159daaf2938aca6e9385209b4214d25b747
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Wed Jan 26 16:05:16 2022 +0800
Fix batch message ack does not decrease the unacked-msg count. (#13383)
(cherry picked from commit d64f2916e0feee1b09459b4db6094db1143067a1)
---
.../org/apache/pulsar/broker/service/Consumer.java | 129 ++++++++++++++++++---
.../pulsar/broker/service/BatchMessageTest.java | 2 +-
.../BatchMessageWithBatchIndexLevelTest.java | 108 +++++++++++++++++
.../PersistentDispatcherFailoverConsumerTest.java | 79 ++++++-------
4 files changed, 265 insertions(+), 53 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
index 4f886d3..cf9f5bf 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -24,6 +24,7 @@ import com.google.common.collect.Lists;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.Promise;
import java.util.ArrayList;
+import java.util.BitSet;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -56,6 +57,7 @@ import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.stats.Rate;
import org.apache.pulsar.common.util.DateFormatter;
import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.collections.BitSetRecyclable;
import org.apache.pulsar.transaction.common.exception.TransactionConflictException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -126,6 +128,7 @@ public class Consumer {
private PositionImpl readPositionWhenJoining;
private final String clientAddress; // IP address only, no port number included
private final MessageId startMessageId;
+ private final boolean isAcknowledgmentAtBatchIndexLevelEnabled;
public Consumer(Subscription subscription, SubType subType, String topicName, long consumerId,
int priorityLevel, String consumerName,
@@ -182,6 +185,8 @@ public class Consumer {
}
this.clientAddress = cnx.clientSourceAddress();
+ this.isAcknowledgmentAtBatchIndexLevelEnabled = subscription.getTopic().getBrokerService()
+ .getPulsar().getConfiguration().isAcknowledgmentAtBatchIndexLevelEnabled();
}
public SubType subType() {
@@ -233,7 +238,7 @@ public class Consumer {
writePromise.setSuccess(null);
return writePromise;
}
-
+ int unackedMessages = totalMessages;
// Note
// Must ensure that the message is written to the pendingAcks before sent is first , because this consumer
// is possible to disconnect at this time.
@@ -243,11 +248,16 @@ public class Consumer {
if (entry != null) {
int batchSize = batchSizes.getBatchSize(i);
int stickyKeyHash = getStickyKeyHash(entry);
+ long[] ackSet = getCursorAckSet(PositionImpl.get(entry.getLedgerId(), entry.getEntryId()));
+ if (ackSet != null) {
+ unackedMessages -= (batchSize - BitSet.valueOf(ackSet).cardinality());
+ }
pendingAcks.put(entry.getLedgerId(), entry.getEntryId(), batchSize, stickyKeyHash);
- if (log.isDebugEnabled()){
+ if (log.isDebugEnabled()) {
log.debug("[{}-{}] Added {}:{} ledger entry with batchSize of {} to pendingAcks in"
+ " broker.service.Consumer for consumerId: {}",
- topicName, subscription, entry.getLedgerId(), entry.getEntryId(), batchSize, consumerId);
+ topicName, subscription, entry.getLedgerId(), entry.getEntryId(), batchSize,
+ consumerId);
}
}
}
@@ -267,7 +277,7 @@ public class Consumer {
+ " for consumerId: {}; avgMessagesPerEntry is {}",
topicName, subscription, ackedCount, totalMessages, consumerId, tmpAvgMessagesPerEntry);
}
- incrementUnackedMessages(totalMessages);
+ incrementUnackedMessages(unackedMessages);
msgOut.recordMultipleEvents(totalMessages, totalBytes);
msgOutCounter.add(totalMessages);
bytesOutCounter.add(totalBytes);
@@ -381,16 +391,19 @@ public class Consumer {
//this method is for individual ack not carry the transaction
private CompletableFuture<Void> individualAckNormal(CommandAck ack, Map<String, Long> properties) {
List<Position> positionsAcked = new ArrayList<>();
-
for (int i = 0; i < ack.getMessageIdsCount(); i++) {
MessageIdData msgId = ack.getMessageIdAt(i);
PositionImpl position;
+ long ackedCount = 0;
+ long batchSize = getBatchSize(msgId);
+ Consumer ackOwnerConsumer = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId());
if (msgId.getAckSetsCount() > 0) {
long[] ackSets = new long[msgId.getAckSetsCount()];
for (int j = 0; j < msgId.getAckSetsCount(); j++) {
ackSets[j] = msgId.getAckSetAt(j);
}
position = PositionImpl.get(msgId.getLedgerId(), msgId.getEntryId(), ackSets);
+ ackedCount = getAckedCountForBatchIndexLevelEnabled(position, batchSize, ackSets);
if (isTransactionEnabled()) {
//sync the batch position bit set point, in order to delete the position in pending acks
if (Subscription.isIndividualAckMode(subType)) {
@@ -400,7 +413,20 @@ public class Consumer {
}
} else {
position = PositionImpl.get(msgId.getLedgerId(), msgId.getEntryId());
+ if (isAcknowledgmentAtBatchIndexLevelEnabled) {
+ long[] cursorAckSet = getCursorAckSet(position);
+ if (cursorAckSet != null) {
+ ackedCount = batchSize - BitSet.valueOf(cursorAckSet).cardinality();
+ } else {
+ ackedCount = batchSize;
+ }
+ } else {
+ ackedCount = batchSize;
+ }
}
+
+ addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount);
+
positionsAcked.add(position);
checkCanRemovePendingAcksAndHandle(position, msgId);
@@ -430,7 +456,6 @@ public class Consumer {
private CompletableFuture<Void> individualAckWithTransaction(CommandAck ack) {
// Individual ack
List<MutablePair<PositionImpl, Integer>> positionsAcked = new ArrayList<>();
-
if (!isTransactionEnabled()) {
return FutureUtil.failedFuture(
new BrokerServiceException.NotAllowedException("Server don't support transaction ack!"));
@@ -476,6 +501,56 @@ public class Consumer {
return completableFuture;
}
+ private long getBatchSize(MessageIdData msgId) {
+ long batchSize = 1;
+ if (Subscription.isIndividualAckMode(subType)) {
+ LongPair longPair = pendingAcks.get(msgId.getLedgerId(), msgId.getEntryId());
+ // Consumer may ack the msg that not belongs to it.
+ if (longPair == null) {
+ Consumer ackOwnerConsumer = getAckOwnerConsumer(msgId.getLedgerId(), msgId.getEntryId());
+ longPair = ackOwnerConsumer.getPendingAcks().get(msgId.getLedgerId(), msgId.getEntryId());
+ if (longPair != null) {
+ batchSize = longPair.first;
+ }
+ } else {
+ batchSize = longPair.first;
+ }
+ }
+ return batchSize;
+ }
+
+ private long getAckedCountForBatchIndexLevelEnabled(PositionImpl position, long batchSize, long[] ackSets) {
+ long ackedCount = 0;
+ if (isAcknowledgmentAtBatchIndexLevelEnabled) {
+ long[] cursorAckSet = getCursorAckSet(position);
+ if (cursorAckSet != null) {
+ BitSetRecyclable cursorBitSet = BitSetRecyclable.create().resetWords(cursorAckSet);
+ int lastCardinality = cursorBitSet.cardinality();
+ BitSetRecyclable givenBitSet = BitSetRecyclable.create().resetWords(ackSets);
+ cursorBitSet.and(givenBitSet);
+ givenBitSet.recycle();
+ int currentCardinality = cursorBitSet.cardinality();
+ ackedCount = lastCardinality - currentCardinality;
+ cursorBitSet.recycle();
+ } else if (pendingAcks.get(position.getLedgerId(), position.getEntryId()) != null) {
+ ackedCount = batchSize - BitSet.valueOf(ackSets).cardinality();
+ }
+ }
+ return ackedCount;
+ }
+
+ private long getUnAckedCountForBatchIndexLevelEnabled(PositionImpl position, long batchSize) {
+ long unAckedCount = batchSize;
+ if (isAcknowledgmentAtBatchIndexLevelEnabled) {
+ long[] cursorAckSet = getCursorAckSet(position);
+ if (cursorAckSet != null) {
+ BitSetRecyclable cursorBitSet = BitSetRecyclable.create().resetWords(cursorAckSet);
+ unAckedCount = cursorBitSet.cardinality();
+ }
+ }
+ return unAckedCount;
+ }
+
private void checkAckValidationError(CommandAck ack, PositionImpl position) {
if (ack.hasValidationError()) {
log.error("[{}] [{}] Received ack for corrupted message at {} - Reason: {}", subscription,
@@ -489,6 +564,26 @@ public class Consumer {
}
}
+ private Consumer getAckOwnerConsumer(long ledgerId, long entryId) {
+ Consumer ackOwnerConsumer = this;
+ if (Subscription.isIndividualAckMode(subType)) {
+ for (Consumer consumer : subscription.getConsumers()) {
+ if (consumer != this && consumer.getPendingAcks().containsKey(ledgerId, entryId)) {
+ ackOwnerConsumer = consumer;
+ break;
+ }
+ }
+ }
+ return ackOwnerConsumer;
+ }
+
+ private long[] getCursorAckSet(PositionImpl position) {
+ if (!(subscription instanceof PersistentSubscription)) {
+ return null;
+ }
+ return (((PersistentSubscription) subscription).getCursor()).getDeletedBatchIndexesAsLongArray(position);
+ }
+
private boolean isTransactionEnabled() {
return subscription instanceof PersistentSubscription
&& ((PersistentTopic) subscription.getTopic())
@@ -711,7 +806,6 @@ public class Consumer {
? ackOwnedConsumer.getPendingAcks().get(position.getLedgerId(), position.getEntryId())
: null;
if (ackedPosition != null) {
- int totalAckedMsgs = (int) ackedPosition.first;
if (!ackOwnedConsumer.getPendingAcks().remove(position.getLedgerId(), position.getEntryId())) {
// Message was already removed by the other consumer
return;
@@ -721,7 +815,7 @@ public class Consumer {
}
// unblock consumer-throttling when limit check is disabled or receives half of maxUnackedMessages =>
// consumer can start again consuming messages
- int unAckedMsgs = addAndGetUnAckedMsgs(ackOwnedConsumer, -totalAckedMsgs);
+ int unAckedMsgs = UNACKED_MESSAGES_UPDATER.get(ackOwnedConsumer);
if ((((unAckedMsgs <= maxUnackedMessages / 2) && ackOwnedConsumer.blockedConsumerOnUnackedMsgs)
&& ackOwnedConsumer.shouldBlockConsumerOnUnackMsgs())
|| !shouldBlockConsumerOnUnackMsgs()) {
@@ -751,7 +845,9 @@ public class Consumer {
List<PositionImpl> pendingPositions = new ArrayList<>((int) pendingAcks.size());
MutableInt totalRedeliveryMessages = new MutableInt(0);
pendingAcks.forEach((ledgerId, entryId, batchSize, stickyKeyHash) -> {
- totalRedeliveryMessages.add((int) batchSize);
+ int unAckedCount = (int) getUnAckedCountForBatchIndexLevelEnabled(PositionImpl.get(ledgerId, entryId),
+ batchSize);
+ totalRedeliveryMessages.add(unAckedCount);
pendingPositions.add(new PositionImpl(ledgerId, entryId));
});
@@ -775,9 +871,9 @@ public class Consumer {
PositionImpl position = PositionImpl.get(msg.getLedgerId(), msg.getEntryId());
LongPair longPair = pendingAcks.get(position.getLedgerId(), position.getEntryId());
if (longPair != null) {
- long batchSize = longPair.first;
+ int unAckedCount = (int) getUnAckedCountForBatchIndexLevelEnabled(position, longPair.first);
pendingAcks.remove(position.getLedgerId(), position.getEntryId());
- totalRedeliveryMessages += batchSize;
+ totalRedeliveryMessages += unAckedCount;
pendingPositions.add(position);
}
}
@@ -811,8 +907,15 @@ public class Consumer {
}
private int addAndGetUnAckedMsgs(Consumer consumer, int ackedMessages) {
- subscription.addUnAckedMessages(ackedMessages);
- return UNACKED_MESSAGES_UPDATER.addAndGet(consumer, ackedMessages);
+ int unackedMsgs = 0;
+ if (Subscription.isIndividualAckMode(subType)) {
+ subscription.addUnAckedMessages(ackedMessages);
+ unackedMsgs = UNACKED_MESSAGES_UPDATER.addAndGet(consumer, ackedMessages);
+ }
+ if (unackedMsgs < 0) {
+ log.error("unackedMsgs is : {}, ackedMessages : {}, consumer : {}", unackedMsgs, ackedMessages, consumer);
+ }
+ return unackedMsgs;
}
private void clearUnAckedMsgs() {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java
index 1bf27ac..c77d2bf 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java
@@ -668,7 +668,7 @@ public class BatchMessageTest extends BrokerTestBase {
*
* @throws Exception
*/
- @Test(dataProvider = "containerBuilder", timeOut = 3000)
+ @Test(dataProvider = "containerBuilder")
public void testConcurrentBatchMessageAck(BatcherBuilder builder) throws Exception {
int numMsgs = 10;
final String topicName = "persistent://prop/ns-abc/testConcurrentAck-" + UUID.randomUUID();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
new file mode 100644
index 0000000..20ba4f1
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageWithBatchIndexLevelTest.java
@@ -0,0 +1,108 @@
+/**
+ * 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.
+ */
+package org.apache.pulsar.broker.service;
+
+import com.google.common.collect.Lists;
+import lombok.SneakyThrows;
+import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.ConsumerImpl;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.awaitility.Awaitility;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import static org.testng.Assert.assertEquals;
+
+@Test(groups = "broker")
+public class BatchMessageWithBatchIndexLevelTest extends BatchMessageTest {
+
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ conf.setAcknowledgmentAtBatchIndexLevelEnabled(true);
+ super.baseSetup();
+ }
+
+ @Test
+ @SneakyThrows
+ public void testBatchMessageAck() {
+ int numMsgs = 40;
+ final String topicName = "persistent://prop/ns-abc/batchMessageAck-" + UUID.randomUUID();
+ final String subscriptionName = "sub-batch-1";
+
+ ConsumerImpl<byte[]> consumer = (ConsumerImpl<byte[]>) pulsarClient
+ .newConsumer()
+ .topic(topicName)
+ .subscriptionName(subscriptionName)
+ .receiverQueueSize(10)
+ .subscriptionType(SubscriptionType.Shared)
+ .enableBatchIndexAcknowledgment(true)
+ .negativeAckRedeliveryDelay(100, TimeUnit.MILLISECONDS)
+ .subscribe();
+
+ Producer<byte[]> producer = pulsarClient
+ .newProducer()
+ .topic(topicName)
+ .batchingMaxMessages(20)
+ .batchingMaxPublishDelay(1, TimeUnit.HOURS)
+ .enableBatching(true)
+ .create();
+
+ List<CompletableFuture<MessageId>> sendFutureList = Lists.newArrayList();
+ for (int i = 0; i < numMsgs; i++) {
+ byte[] message = ("batch-message-" + i).getBytes();
+ sendFutureList.add(producer.newMessage().value(message).sendAsync());
+ }
+ FutureUtil.waitForAll(sendFutureList).get();
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).get();
+ PersistentDispatcherMultipleConsumers dispatcher = (PersistentDispatcherMultipleConsumers) topic
+ .getSubscription(subscriptionName).getDispatcher();
+ Message<byte[]> receive1 = consumer.receive();
+ Message<byte[]> receive2 = consumer.receive();
+ consumer.acknowledge(receive1);
+ consumer.acknowledge(receive2);
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(dispatcher.getConsumers().get(0).getUnackedMessages(), 18);
+ });
+ Message<byte[]> receive3 = consumer.receive();
+ Message<byte[]> receive4 = consumer.receive();
+ consumer.acknowledge(receive3);
+ consumer.acknowledge(receive4);
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(dispatcher.getConsumers().get(0).getUnackedMessages(), 16);
+ });
+ Message<byte[]> receive5 = consumer.receive();
+ consumer.negativeAcknowledge(receive5);
+ Awaitility.await().pollInterval(1, TimeUnit.MILLISECONDS).untilAsserted(() -> {
+ assertEquals(dispatcher.getConsumers().get(0).getUnackedMessages(), 0);
+ });
+ consumer.receive();
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(dispatcher.getConsumers().get(0).getUnackedMessages(), 16);
+ });
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
index d4aae1a..4b947aa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentDispatcherFailoverConsumerTest.java
@@ -515,15 +515,15 @@ public class PersistentDispatcherFailoverConsumerTest {
PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
- Consumer consumer1 = createConsumer(0, 2, false, 1);
- Consumer consumer2 = createConsumer(0, 2, false, 2);
- Consumer consumer3 = createConsumer(0, 2, false, 3);
- Consumer consumer4 = createConsumer(1, 2, false, 4);
- Consumer consumer5 = createConsumer(1, 1, false, 5);
- Consumer consumer6 = createConsumer(1, 2, false, 6);
- Consumer consumer7 = createConsumer(2, 1, false, 7);
- Consumer consumer8 = createConsumer(2, 1, false, 8);
- Consumer consumer9 = createConsumer(2, 1, false, 9);
+ Consumer consumer1 = createConsumer(topic, 0, 2, false, 1);
+ Consumer consumer2 = createConsumer(topic, 0, 2, false, 2);
+ Consumer consumer3 = createConsumer(topic, 0, 2, false, 3);
+ Consumer consumer4 = createConsumer(topic, 1, 2, false, 4);
+ Consumer consumer5 = createConsumer(topic, 1, 1, false, 5);
+ Consumer consumer6 = createConsumer(topic, 1, 2, false, 6);
+ Consumer consumer7 = createConsumer(topic, 2, 1, false, 7);
+ Consumer consumer8 = createConsumer(topic, 2, 1, false, 8);
+ Consumer consumer9 = createConsumer(topic, 2, 1, false, 9);
dispatcher.addConsumer(consumer1);
dispatcher.addConsumer(consumer2);
dispatcher.addConsumer(consumer3);
@@ -547,7 +547,7 @@ public class PersistentDispatcherFailoverConsumerTest {
Assert.assertEquals(getNextConsumer(dispatcher), consumer7);
Assert.assertEquals(getNextConsumer(dispatcher), consumer8);
// in between add upper priority consumer with more permits
- Consumer consumer10 = createConsumer(0, 2, false, 10);
+ Consumer consumer10 = createConsumer(topic, 0, 2, false, 10);
dispatcher.addConsumer(consumer10);
Assert.assertEquals(getNextConsumer(dispatcher), consumer10);
Assert.assertEquals(getNextConsumer(dispatcher), consumer10);
@@ -559,12 +559,12 @@ public class PersistentDispatcherFailoverConsumerTest {
public void testFewBlockedConsumerSamePriority() throws Exception{
PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
- Consumer consumer1 = createConsumer(0, 2, false, 1);
- Consumer consumer2 = createConsumer(0, 2, false, 2);
- Consumer consumer3 = createConsumer(0, 2, false, 3);
- Consumer consumer4 = createConsumer(0, 2, false, 4);
- Consumer consumer5 = createConsumer(0, 1, true, 5);
- Consumer consumer6 = createConsumer(0, 2, true, 6);
+ Consumer consumer1 = createConsumer(topic, 0, 2, false, 1);
+ Consumer consumer2 = createConsumer(topic, 0, 2, false, 2);
+ Consumer consumer3 = createConsumer(topic, 0, 2, false, 3);
+ Consumer consumer4 = createConsumer(topic, 0, 2, false, 4);
+ Consumer consumer5 = createConsumer(topic, 0, 1, true, 5);
+ Consumer consumer6 = createConsumer(topic, 0, 2, true, 6);
dispatcher.addConsumer(consumer1);
dispatcher.addConsumer(consumer2);
dispatcher.addConsumer(consumer3);
@@ -586,18 +586,18 @@ public class PersistentDispatcherFailoverConsumerTest {
public void testFewBlockedConsumerDifferentPriority() throws Exception {
PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
- Consumer consumer1 = createConsumer(0, 2, false, 1);
- Consumer consumer2 = createConsumer(0, 2, false, 2);
- Consumer consumer3 = createConsumer(0, 2, false, 3);
- Consumer consumer4 = createConsumer(0, 2, false, 4);
- Consumer consumer5 = createConsumer(0, 1, true, 5);
- Consumer consumer6 = createConsumer(0, 2, true, 6);
- Consumer consumer7 = createConsumer(1, 2, false, 7);
- Consumer consumer8 = createConsumer(1, 10, true, 8);
- Consumer consumer9 = createConsumer(1, 2, false, 9);
- Consumer consumer10 = createConsumer(2, 2, false, 10);
- Consumer consumer11 = createConsumer(2, 10, true, 11);
- Consumer consumer12 = createConsumer(2, 2, false, 12);
+ Consumer consumer1 = createConsumer(topic, 0, 2, false, 1);
+ Consumer consumer2 = createConsumer(topic, 0, 2, false, 2);
+ Consumer consumer3 = createConsumer(topic, 0, 2, false, 3);
+ Consumer consumer4 = createConsumer(topic, 0, 2, false, 4);
+ Consumer consumer5 = createConsumer(topic, 0, 1, true, 5);
+ Consumer consumer6 = createConsumer(topic, 0, 2, true, 6);
+ Consumer consumer7 = createConsumer(topic, 1, 2, false, 7);
+ Consumer consumer8 = createConsumer(topic, 1, 10, true, 8);
+ Consumer consumer9 = createConsumer(topic, 1, 2, false, 9);
+ Consumer consumer10 = createConsumer(topic, 2, 2, false, 10);
+ Consumer consumer11 = createConsumer(topic, 2, 10, true, 11);
+ Consumer consumer12 = createConsumer(topic, 2, 2, false, 12);
dispatcher.addConsumer(consumer1);
dispatcher.addConsumer(consumer2);
dispatcher.addConsumer(consumer3);
@@ -625,8 +625,8 @@ public class PersistentDispatcherFailoverConsumerTest {
Assert.assertEquals(getNextConsumer(dispatcher), consumer10);
Assert.assertEquals(getNextConsumer(dispatcher), consumer12);
// add consumer with lower priority again
- Consumer consumer13 = createConsumer(0, 2, false, 13);
- Consumer consumer14 = createConsumer(0, 2, true, 14);
+ Consumer consumer13 = createConsumer(topic, 0, 2, false, 13);
+ Consumer consumer14 = createConsumer(topic, 0, 2, true, 14);
dispatcher.addConsumer(consumer13);
dispatcher.addConsumer(consumer14);
Assert.assertEquals(getNextConsumer(dispatcher), consumer13);
@@ -640,13 +640,13 @@ public class PersistentDispatcherFailoverConsumerTest {
public void testFewBlockedConsumerDifferentPriority2() throws Exception {
PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
PersistentDispatcherMultipleConsumers dispatcher = new PersistentDispatcherMultipleConsumers(topic, cursorMock, null);
- Consumer consumer1 = createConsumer(0, 2, true, 1);
- Consumer consumer2 = createConsumer(0, 2, true, 2);
- Consumer consumer3 = createConsumer(0, 2, true, 3);
- Consumer consumer4 = createConsumer(1, 2, false, 4);
- Consumer consumer5 = createConsumer(1, 1, false, 5);
- Consumer consumer6 = createConsumer(2, 1, false, 6);
- Consumer consumer7 = createConsumer(2, 2, true, 7);
+ Consumer consumer1 = createConsumer(topic, 0, 2, true, 1);
+ Consumer consumer2 = createConsumer(topic, 0, 2, true, 2);
+ Consumer consumer3 = createConsumer(topic, 0, 2, true, 3);
+ Consumer consumer4 = createConsumer(topic, 1, 2, false, 4);
+ Consumer consumer5 = createConsumer(topic, 1, 1, false, 5);
+ Consumer consumer6 = createConsumer(topic, 2, 1, false, 6);
+ Consumer consumer7 = createConsumer(topic, 2, 2, true, 7);
dispatcher.addConsumer(consumer1);
dispatcher.addConsumer(consumer2);
dispatcher.addConsumer(consumer3);
@@ -676,9 +676,10 @@ public class PersistentDispatcherFailoverConsumerTest {
return null;
}
- private Consumer createConsumer(int priority, int permit, boolean blocked, int id) throws Exception {
+ private Consumer createConsumer(PersistentTopic topic, int priority, int permit, boolean blocked, int id) throws Exception {
+ PersistentSubscription sub = new PersistentSubscription(topic, "sub-1", cursorMock, false);
Consumer consumer =
- new Consumer(null, SubType.Shared, "test-topic", id, priority, ""+id, 5000,
+ new Consumer(sub, SubType.Shared, "test-topic", id, priority, ""+id, 5000,
serverCnx, "appId", Collections.emptyMap(), false /* read compacted */, InitialPosition.Latest, null, MessageId.latest);
try {
consumer.flowPermits(permit);
[pulsar] 07/11: NPE when get isAllowAutoUploadSchema (#13831)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 76fac617f57130795c86f01e3d0e06ff1a2c99a1
Author: gaozhangmin <ga...@qq.com>
AuthorDate: Fri Jan 21 20:28:56 2022 +0800
NPE when get isAllowAutoUploadSchema (#13831)
### Motivation
there were some potential NPE bug when get null value by pulsar-admin
```
Caused by: java.lang.NullPointerException
at org.apache.pulsar.broker.admin.impl.NamespacesBase.internalGetIsAllowAutoUpdateSchema(NamespacesBase.java:2423) ~[org.apache.pulsar-pulsar-broker-2.9.1.jar:2.9.1]
at org.apache.pulsar.broker.admin.v2.Namespaces.getIsAllowAutoUpdateSchema(Namespaces.java:1631) ~[org.apache.pulsar-pulsar-broker-2.9.1.jar:2.9.1]
at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
```
### Modifications
1、set default value true to is_allow_auto_update_schema in Policies as same as ServiceConfiguration
2、changes from boolean to Boolean, long to Long, prevent potential NPE.
(cherry picked from commit 0a8794b768b414f8e5bc9b3abb76e57de9455add)
---
.../main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java | 3 +++
1 file changed, 3 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index 3d67008..c2ad1d8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -2411,6 +2411,9 @@ public abstract class NamespacesBase extends AdminResource {
protected boolean internalGetIsAllowAutoUpdateSchema() {
validateNamespacePolicyOperation(namespaceName, PolicyName.SCHEMA_COMPATIBILITY_STRATEGY,
PolicyOperation.READ);
+ if (getNamespacePolicies(namespaceName).is_allow_auto_update_schema == null) {
+ return pulsar().getConfig().isAllowAutoUpdateSchemaEnabled();
+ }
return getNamespacePolicies(namespaceName).is_allow_auto_update_schema;
}
[pulsar] 10/11: [Broker] Fix read schema compatibility strategy priority (#13938)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9601de1a2d95e0df5d51a2c9c44ed578ae0668e0
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Wed Jan 26 15:46:18 2022 +0800
[Broker] Fix read schema compatibility strategy priority (#13938)
Signed-off-by: Zixuan Liu <no...@gmail.com>
When we defined the `schemaCompatibilityStrategy=ALWAYS_INCOMPATIBLE` in `broker.conf`, and a namespace policies so like:
```
schema_auto_update_compatibility_strategy = SchemaAutoUpdateCompatibilityStrategy.Full
schema_compatibility_strategy = null
```
We should get `SchemaCompatibilityStrategy.FULL` by `pulsar-admin namespaces get-schema-compatibility-strategy <ns>`, but got `SchemaCompatibilityStrategy.ALWAYS_INCOMPATIBLE`, this is incorrect response.
- Use `null` as `Policies#schema_auto_update_compatibility_strategy` value
- Use `FULL` as `schemaCompatibilityStrategy` value in `broker.conf` and update configuration code
- Only return `Policies#schema_compatibility_strategy` when get schema compatibility strategy of namespace
- Change the read schema compatibility strategy priority
(cherry picked from commit c18d64526708c6dd051217feba3c31a73507620c)
---
conf/broker.conf | 8 ++--
.../apache/pulsar/broker/ServiceConfiguration.java | 11 +++--
.../apache/pulsar/broker/admin/AdminResource.java | 15 ++++++
.../pulsar/broker/admin/impl/NamespacesBase.java | 11 +----
.../broker/admin/impl/SchemasResourceBase.java | 42 +++++------------
.../pulsar/broker/service/AbstractTopic.java | 18 ++++----
.../broker/admin/AdminApiSchemaAutoUpdateTest.java | 22 +++------
.../pulsar/broker/admin/AdminApiSchemaTest.java | 53 ++++++++++++++++++++++
.../SchemaCompatibilityCheckTest.java | 6 +--
.../SchemaTypeCompatibilityCheckTest.java | 8 ++--
.../pulsar/common/policies/data/Policies.java | 3 +-
.../policies/data/SchemaCompatibilityStrategy.java | 5 +-
12 files changed, 121 insertions(+), 81 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index a51a5f7..ae5dd2e 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1215,12 +1215,10 @@ schemaRegistryStorageClassName=org.apache.pulsar.broker.service.schema.Bookkeepe
# if you enable this setting, it will cause non-java clients failed to produce.
isSchemaValidationEnforced=false
-# The schema compatibility strategy in broker level. If this config in namespace policy is `UNDEFINED`,
-# broker will use it in broker level. If schemaCompatibilityStrategy is `UNDEFINED` will use `FULL`.
-# SchemaCompatibilityStrategy : UNDEFINED, ALWAYS_INCOMPATIBLE, ALWAYS_COMPATIBLE, BACKWARD, FORWARD,
+# The schema compatibility strategy in broker level.
+# SchemaCompatibilityStrategy : ALWAYS_INCOMPATIBLE, ALWAYS_COMPATIBLE, BACKWARD, FORWARD,
# FULL, BACKWARD_TRANSITIVE, FORWARD_TRANSITIVE, FULL_TRANSITIVE
-# default : UNDEFINED
-schemaCompatibilityStrategy=
+schemaCompatibilityStrategy=FULL
### --- Ledger Offloading --- ###
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 21c0e5f..904fcec 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -1971,10 +1971,9 @@ public class ServiceConfiguration implements PulsarConfiguration {
@FieldContext(
category = CATEGORY_SCHEMA,
- doc = "The schema compatibility strategy in broker level. If this config in namespace policy is `UNDEFINED`"
- + ", schema compatibility strategy check will use it in broker level."
+ doc = "The schema compatibility strategy in broker level"
)
- private SchemaCompatibilityStrategy schemaCompatibilityStrategy = SchemaCompatibilityStrategy.UNDEFINED;
+ private SchemaCompatibilityStrategy schemaCompatibilityStrategy = SchemaCompatibilityStrategy.FULL;
/**** --- WebSocket --- ****/
@FieldContext(
@@ -2396,4 +2395,10 @@ public class ServiceConfiguration implements PulsarConfiguration {
}
}
+ public SchemaCompatibilityStrategy getSchemaCompatibilityStrategy() {
+ if (SchemaCompatibilityStrategy.isUndefined(schemaCompatibilityStrategy)) {
+ return SchemaCompatibilityStrategy.FULL;
+ }
+ return schemaCompatibilityStrategy;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index 6e601af..5a31524 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -57,6 +57,7 @@ import org.apache.pulsar.common.policies.data.NamespaceOperation;
import org.apache.pulsar.common.policies.data.PersistencePolicies;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.common.policies.data.SubscribeRate;
import org.apache.pulsar.common.policies.data.TopicOperation;
import org.apache.pulsar.common.policies.data.TopicPolicies;
@@ -798,6 +799,20 @@ public abstract class AdminResource extends PulsarWebResource {
}
}
+ protected CompletableFuture<SchemaCompatibilityStrategy> getSchemaCompatibilityStrategyAsync() {
+ return getNamespacePoliciesAsync(namespaceName).thenApply(policies -> {
+ SchemaCompatibilityStrategy schemaCompatibilityStrategy = policies.schema_compatibility_strategy;
+ if (SchemaCompatibilityStrategy.isUndefined(schemaCompatibilityStrategy)) {
+ schemaCompatibilityStrategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy(
+ policies.schema_auto_update_compatibility_strategy);
+ if (SchemaCompatibilityStrategy.isUndefined(schemaCompatibilityStrategy)) {
+ schemaCompatibilityStrategy = pulsar().getConfig().getSchemaCompatibilityStrategy();
+ }
+ }
+ return schemaCompatibilityStrategy;
+ });
+ }
+
@CanIgnoreReturnValue
public static <T> T checkNotNull(T reference) {
return com.google.common.base.Preconditions.checkNotNull(reference);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index c2ad1d8..9b45cb0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -2349,15 +2349,8 @@ public abstract class NamespacesBase extends AdminResource {
validateNamespacePolicyOperation(namespaceName, PolicyName.SCHEMA_COMPATIBILITY_STRATEGY,
PolicyOperation.READ);
Policies policies = getNamespacePolicies(namespaceName);
- SchemaCompatibilityStrategy schemaCompatibilityStrategy = policies.schema_compatibility_strategy;
- if (schemaCompatibilityStrategy == SchemaCompatibilityStrategy.UNDEFINED) {
- schemaCompatibilityStrategy = pulsar().getConfig().getSchemaCompatibilityStrategy();
- if (schemaCompatibilityStrategy == SchemaCompatibilityStrategy.UNDEFINED) {
- schemaCompatibilityStrategy = SchemaCompatibilityStrategy
- .fromAutoUpdatePolicy(policies.schema_auto_update_compatibility_strategy);
- }
- }
- return schemaCompatibilityStrategy;
+
+ return policies.schema_compatibility_strategy;
}
@Deprecated
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java
index d2decdd..a4c3f5e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java
@@ -37,8 +37,6 @@ import org.apache.pulsar.broker.service.schema.exceptions.InvalidSchemaDataExcep
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.client.internal.DefaultImplementation;
import org.apache.pulsar.common.naming.TopicName;
-import org.apache.pulsar.common.policies.data.Policies;
-import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.common.protocol.schema.DeleteSchemaResponse;
import org.apache.pulsar.common.protocol.schema.GetAllVersionsSchemaResponse;
import org.apache.pulsar.common.protocol.schema.GetSchemaResponse;
@@ -135,16 +133,7 @@ public class SchemasResourceBase extends AdminResource {
public void postSchema(PostSchemaPayload payload, boolean authoritative, AsyncResponse response) {
validateDestinationAndAdminOperation(authoritative);
- getNamespacePoliciesAsync(namespaceName).thenAccept(policies -> {
- SchemaCompatibilityStrategy schemaCompatibilityStrategy = policies.schema_compatibility_strategy;
- if (schemaCompatibilityStrategy == SchemaCompatibilityStrategy.UNDEFINED) {
- schemaCompatibilityStrategy =
- pulsar().getConfig().getSchemaCompatibilityStrategy();
- if (schemaCompatibilityStrategy == SchemaCompatibilityStrategy.UNDEFINED) {
- schemaCompatibilityStrategy = SchemaCompatibilityStrategy
- .fromAutoUpdatePolicy(policies.schema_auto_update_compatibility_strategy);
- }
- }
+ getSchemaCompatibilityStrategyAsync().thenAccept(schemaCompatibilityStrategy -> {
byte[] data;
if (SchemaType.KEY_VALUE.name().equals(payload.getType())) {
data = DefaultImplementation
@@ -192,26 +181,17 @@ public class SchemasResourceBase extends AdminResource {
validateDestinationAndAdminOperation(authoritative);
String schemaId = getSchemaId();
- Policies policies = getNamespacePolicies(namespaceName);
-
- SchemaCompatibilityStrategy schemaCompatibilityStrategy;
- if (policies.schema_compatibility_strategy == SchemaCompatibilityStrategy.UNDEFINED) {
- schemaCompatibilityStrategy = SchemaCompatibilityStrategy
- .fromAutoUpdatePolicy(policies.schema_auto_update_compatibility_strategy);
- } else {
- schemaCompatibilityStrategy = policies.schema_compatibility_strategy;
- }
- pulsar().getSchemaRegistryService()
- .isCompatible(schemaId,
- SchemaData.builder().data(payload.getSchema().getBytes(Charsets.UTF_8)).isDeleted(false)
- .timestamp(clock.millis()).type(SchemaType.valueOf(payload.getType()))
- .user(defaultIfEmpty(clientAppId(), "")).props(payload.getProperties()).build(),
- schemaCompatibilityStrategy)
- .thenAccept(isCompatible -> response.resume(Response.accepted()
- .entity(IsCompatibilityResponse.builder().isCompatibility(isCompatible)
- .schemaCompatibilityStrategy(schemaCompatibilityStrategy.name()).build())
- .build()))
+ getSchemaCompatibilityStrategyAsync().thenCompose(schemaCompatibilityStrategy -> pulsar()
+ .getSchemaRegistryService().isCompatible(schemaId,
+ SchemaData.builder().data(payload.getSchema().getBytes(Charsets.UTF_8)).isDeleted(false)
+ .timestamp(clock.millis()).type(SchemaType.valueOf(payload.getType()))
+ .user(defaultIfEmpty(clientAppId(), "")).props(payload.getProperties()).build(),
+ schemaCompatibilityStrategy)
+ .thenAccept(isCompatible -> response.resume(Response.accepted()
+ .entity(IsCompatibilityResponse.builder().isCompatibility(isCompatible)
+ .schemaCompatibilityStrategy(schemaCompatibilityStrategy.name()).build())
+ .build())))
.exceptionally(error -> {
response.resume(new RestException(error));
return null;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 65296a7..d93a20a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -540,17 +540,19 @@ public abstract class AbstractTopic implements Topic {
if (SystemTopicClient.isSystemTopic(TopicName.get(this.topic))) {
schemaCompatibilityStrategy =
brokerService.pulsar().getConfig().getSystemTopicSchemaCompatibilityStrategy();
- } else if (policies.schema_compatibility_strategy == SchemaCompatibilityStrategy.UNDEFINED) {
- schemaCompatibilityStrategy = brokerService.pulsar()
- .getConfig().getSchemaCompatibilityStrategy();
- if (schemaCompatibilityStrategy == SchemaCompatibilityStrategy.UNDEFINED) {
- schemaCompatibilityStrategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy(
- policies.schema_auto_update_compatibility_strategy);
+ return;
+ }
+
+ schemaCompatibilityStrategy = policies.schema_compatibility_strategy;
+ if (SchemaCompatibilityStrategy.isUndefined(schemaCompatibilityStrategy)) {
+ schemaCompatibilityStrategy = SchemaCompatibilityStrategy.fromAutoUpdatePolicy(
+ policies.schema_auto_update_compatibility_strategy);
+ if (SchemaCompatibilityStrategy.isUndefined(schemaCompatibilityStrategy)) {
+ schemaCompatibilityStrategy = brokerService.pulsar().getConfig().getSchemaCompatibilityStrategy();
}
- } else {
- schemaCompatibilityStrategy = policies.schema_compatibility_strategy;
}
}
+
private static final Summary PUBLISH_LATENCY = Summary.build("pulsar_broker_publish_latency", "-")
.quantile(0.0)
.quantile(0.50)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaAutoUpdateTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaAutoUpdateTest.java
index ec2b1e8..0522366 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaAutoUpdateTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaAutoUpdateTest.java
@@ -33,12 +33,8 @@ import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.common.policies.data.ClusterDataImpl;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.SchemaAutoUpdateCompatibilityStrategy;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -47,9 +43,6 @@ import org.testng.annotations.Test;
@Slf4j
@Test(groups = "broker")
public class AdminApiSchemaAutoUpdateTest extends MockedPulsarServiceBaseTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(AdminApiSchemaAutoUpdateTest.class);
-
@BeforeMethod
@Override
public void setup() throws Exception {
@@ -72,8 +65,8 @@ public class AdminApiSchemaAutoUpdateTest extends MockedPulsarServiceBaseTest {
}
private void testAutoUpdateBackward(String namespace, String topicName) throws Exception {
- Assert.assertEquals(admin.namespaces().getSchemaAutoUpdateCompatibilityStrategy(namespace),
- SchemaAutoUpdateCompatibilityStrategy.Full);
+ Assert.assertNull(admin.namespaces().getSchemaAutoUpdateCompatibilityStrategy(namespace));
+
admin.namespaces().setSchemaAutoUpdateCompatibilityStrategy(namespace,
SchemaAutoUpdateCompatibilityStrategy.Backward);
@@ -96,8 +89,8 @@ public class AdminApiSchemaAutoUpdateTest extends MockedPulsarServiceBaseTest {
}
private void testAutoUpdateForward(String namespace, String topicName) throws Exception {
- Assert.assertEquals(admin.namespaces().getSchemaAutoUpdateCompatibilityStrategy(namespace),
- SchemaAutoUpdateCompatibilityStrategy.Full);
+ Assert.assertNull(admin.namespaces().getSchemaAutoUpdateCompatibilityStrategy(namespace));
+
admin.namespaces().setSchemaAutoUpdateCompatibilityStrategy(namespace,
SchemaAutoUpdateCompatibilityStrategy.Forward);
@@ -119,8 +112,7 @@ public class AdminApiSchemaAutoUpdateTest extends MockedPulsarServiceBaseTest {
}
private void testAutoUpdateFull(String namespace, String topicName) throws Exception {
- Assert.assertEquals(admin.namespaces().getSchemaAutoUpdateCompatibilityStrategy(namespace),
- SchemaAutoUpdateCompatibilityStrategy.Full);
+ Assert.assertNull(admin.namespaces().getSchemaAutoUpdateCompatibilityStrategy(namespace));
try (Producer<V1Data> p = pulsarClient.newProducer(Schema.AVRO(V1Data.class)).topic(topicName).create()) {
p.send(new V1Data("test1", 1));
@@ -147,8 +139,8 @@ public class AdminApiSchemaAutoUpdateTest extends MockedPulsarServiceBaseTest {
}
private void testAutoUpdateDisabled(String namespace, String topicName) throws Exception {
- Assert.assertEquals(admin.namespaces().getSchemaAutoUpdateCompatibilityStrategy(namespace),
- SchemaAutoUpdateCompatibilityStrategy.Full);
+ Assert.assertNull(admin.namespaces().getSchemaAutoUpdateCompatibilityStrategy(namespace));
+
admin.namespaces().setSchemaAutoUpdateCompatibilityStrategy(namespace,
SchemaAutoUpdateCompatibilityStrategy.AutoUpdateDisabled);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java
index dce60e0..64e4251 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.broker.admin;
import static java.lang.String.format;
import static java.nio.charset.StandardCharsets.US_ASCII;
+import static org.junit.Assert.assertNull;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doReturn;
import static org.testng.Assert.assertEquals;
@@ -45,9 +46,11 @@ import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.ClusterDataImpl;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
import org.apache.pulsar.common.policies.data.SchemaAutoUpdateCompatibilityStrategy;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.common.schema.SchemaInfoWithVersion;
+import org.awaitility.Awaitility;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
@@ -61,6 +64,8 @@ import org.testng.annotations.Test;
public class AdminApiSchemaTest extends MockedPulsarServiceBaseTest {
final String cluster = "test";
+ private final String schemaCompatibilityNamespace = "schematest/test-schema-compatibility-ns";
+
@BeforeMethod
@Override
public void setup() throws Exception {
@@ -72,6 +77,7 @@ public class AdminApiSchemaTest extends MockedPulsarServiceBaseTest {
admin.tenants().createTenant("schematest", tenantInfo);
admin.namespaces().createNamespace("schematest/test", Sets.newHashSet("test"));
admin.namespaces().createNamespace("schematest/"+cluster+"/test", Sets.newHashSet("test"));
+ admin.namespaces().createNamespace(schemaCompatibilityNamespace, Sets.newHashSet("test"));
}
@AfterMethod(alwaysRun = true)
@@ -349,4 +355,51 @@ public class AdminApiSchemaTest extends MockedPulsarServiceBaseTest {
assertEquals(ledgerInfo.entries, entryId + 1);
assertEquals(ledgerInfo.size, length);
}
+
+ @Test
+ public void testGetSchemaCompatibilityStrategy() throws PulsarAdminException {
+ assertEquals(admin.namespaces().getSchemaCompatibilityStrategy(schemaCompatibilityNamespace),
+ SchemaCompatibilityStrategy.UNDEFINED);
+ }
+
+ @Test
+ public void testGetSchemaAutoUpdateCompatibilityStrategy() throws PulsarAdminException {
+ assertNull(admin.namespaces().getSchemaAutoUpdateCompatibilityStrategy(schemaCompatibilityNamespace));
+ }
+
+ @Test
+ public void testGetSchemaCompatibilityStrategyWhenSetSchemaAutoUpdateCompatibilityStrategy()
+ throws PulsarAdminException {
+ assertEquals(admin.namespaces().getSchemaCompatibilityStrategy(schemaCompatibilityNamespace),
+ SchemaCompatibilityStrategy.UNDEFINED);
+
+ admin.namespaces().setSchemaAutoUpdateCompatibilityStrategy(schemaCompatibilityNamespace,
+ SchemaAutoUpdateCompatibilityStrategy.Forward);
+ Awaitility.await().untilAsserted(() -> assertEquals(SchemaAutoUpdateCompatibilityStrategy.Forward,
+ admin.namespaces().getSchemaAutoUpdateCompatibilityStrategy(schemaCompatibilityNamespace)
+ ));
+
+ assertEquals(admin.namespaces().getSchemaCompatibilityStrategy(schemaCompatibilityNamespace),
+ SchemaCompatibilityStrategy.UNDEFINED);
+
+ admin.namespaces().setSchemaCompatibilityStrategy(schemaCompatibilityNamespace,
+ SchemaCompatibilityStrategy.BACKWARD);
+ Awaitility.await().untilAsserted(() -> assertEquals(SchemaCompatibilityStrategy.BACKWARD,
+ admin.namespaces().getSchemaCompatibilityStrategy(schemaCompatibilityNamespace)));
+ }
+
+ @Test
+ public void testGetSchemaCompatibilityStrategyWhenSetBrokerLevelAndSchemaAutoUpdateCompatibilityStrategy()
+ throws PulsarAdminException {
+ pulsar.getConfiguration().setSchemaCompatibilityStrategy(SchemaCompatibilityStrategy.FORWARD);
+
+ assertEquals(admin.namespaces().getSchemaCompatibilityStrategy(schemaCompatibilityNamespace),
+ SchemaCompatibilityStrategy.UNDEFINED);
+
+ admin.namespaces().setSchemaAutoUpdateCompatibilityStrategy(schemaCompatibilityNamespace,
+ SchemaAutoUpdateCompatibilityStrategy.AlwaysCompatible);
+ Awaitility.await().untilAsserted(() -> assertEquals(
+ admin.namespaces().getSchemaCompatibilityStrategy(schemaCompatibilityNamespace),
+ SchemaCompatibilityStrategy.UNDEFINED));
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
index 80168b9..5b12f37 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
@@ -240,7 +240,7 @@ public class SchemaCompatibilityCheckTest extends MockedPulsarServiceBaseTest {
);
assertEquals(admin.namespaces().getSchemaCompatibilityStrategy(namespaceName.toString()),
- SchemaCompatibilityStrategy.FULL);
+ SchemaCompatibilityStrategy.UNDEFINED);
admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy);
admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo());
@@ -320,7 +320,7 @@ public class SchemaCompatibilityCheckTest extends MockedPulsarServiceBaseTest {
);
assertEquals(admin.namespaces().getSchemaCompatibilityStrategy(namespaceName.toString()),
- SchemaCompatibilityStrategy.FULL);
+ SchemaCompatibilityStrategy.UNDEFINED);
admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy);
admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo());
@@ -399,7 +399,7 @@ public class SchemaCompatibilityCheckTest extends MockedPulsarServiceBaseTest {
);
assertEquals(admin.namespaces().getSchemaCompatibilityStrategy(namespaceName.toString()),
- SchemaCompatibilityStrategy.FULL);
+ SchemaCompatibilityStrategy.UNDEFINED);
byte[] changeSchemaBytes = (new String(Schema.AVRO(Schemas.PersonOne.class)
.getSchemaInfo().getSchema(), UTF_8) + "/n /n /n").getBytes();
SchemaInfo schemaInfo = SchemaInfo.builder().type(SchemaType.AVRO).schema(changeSchemaBytes).build();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaTypeCompatibilityCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaTypeCompatibilityCheckTest.java
index 345c9c7..45f0076 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaTypeCompatibilityCheckTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaTypeCompatibilityCheckTest.java
@@ -37,8 +37,8 @@ import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.schema.Schemas;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.BeforeClass;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
import java.util.Collections;
@@ -57,7 +57,7 @@ public class SchemaTypeCompatibilityCheckTest extends MockedPulsarServiceBaseTes
private static final String namespace = "test-namespace";
private static final String namespaceName = PUBLIC_TENANT + "/" + namespace;
- @BeforeClass
+ @BeforeMethod
@Override
public void setup() throws Exception {
super.internalSetup();
@@ -73,7 +73,7 @@ public class SchemaTypeCompatibilityCheckTest extends MockedPulsarServiceBaseTes
}
- @AfterClass(alwaysRun = true)
+ @AfterMethod(alwaysRun = true)
@Override
public void cleanup() throws Exception {
super.internalCleanup();
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/Policies.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
index 8622715..43d266f 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/Policies.java
@@ -100,8 +100,7 @@ public class Policies {
@SuppressWarnings("checkstyle:MemberName")
@Deprecated
- public SchemaAutoUpdateCompatibilityStrategy schema_auto_update_compatibility_strategy =
- SchemaAutoUpdateCompatibilityStrategy.Full;
+ public SchemaAutoUpdateCompatibilityStrategy schema_auto_update_compatibility_strategy = null;
@SuppressWarnings("checkstyle:MemberName")
public SchemaCompatibilityStrategy schema_compatibility_strategy = SchemaCompatibilityStrategy.UNDEFINED;
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SchemaCompatibilityStrategy.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SchemaCompatibilityStrategy.java
index 9a4f74c..f3b4569 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SchemaCompatibilityStrategy.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/SchemaCompatibilityStrategy.java
@@ -71,10 +71,13 @@ public enum SchemaCompatibilityStrategy {
FULL_TRANSITIVE;
+ public static boolean isUndefined(SchemaCompatibilityStrategy strategy) {
+ return strategy == null || strategy == SchemaCompatibilityStrategy.UNDEFINED;
+ }
public static SchemaCompatibilityStrategy fromAutoUpdatePolicy(SchemaAutoUpdateCompatibilityStrategy strategy) {
if (strategy == null) {
- return SchemaCompatibilityStrategy.ALWAYS_INCOMPATIBLE;
+ return null;
}
switch (strategy) {
case Backward:
[pulsar] 02/11: Issue 12019 Fix flaky test ManagedLedgerTest.testMaximumRolloverTime (#12030)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5a491c45766a68dd9bd8b76155cbfd5b108b492a
Author: Md Mostafijur Rahman <mo...@gmail.com>
AuthorDate: Tue Jan 18 08:39:20 2022 -0600
Issue 12019 Fix flaky test ManagedLedgerTest.testMaximumRolloverTime (#12030)
(cherry picked from commit ae625ef22e74992c89a459ba732cb0ba6c89605a)
---
.../java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index e378438..6d6cfc9 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -1774,11 +1774,12 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
assertEquals(ledger.getLedgersInfoAsList().size(), 1);
- Thread.sleep(2000);
-
ledger.addEntry("data".getBytes());
ledger.addEntry("data".getBytes());
- assertEquals(ledger.getLedgersInfoAsList().size(), 2);
+
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(ledger.getLedgersInfoAsList().size(), 2);
+ });
}
@Test
[pulsar] 06/11: Getting rid of CVEs in Solr connector (#13822)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 7751597b1ba763d46b4678cb10e78c1187e46476
Author: Andrey Yegorov <86...@users.noreply.github.com>
AuthorDate: Wed Jan 19 23:52:46 2022 -0800
Getting rid of CVEs in Solr connector (#13822)
CVE-2019-17638
CVE-2020-27216
CVE-2020-27218
CVE-2020-27223
CVE-2021-28165
CVE-2021-28169
CVE-2021-34428
CVE-2021-27905
CVE-2021-29262
CVE-2021-29943
CVE-2021-44548
(cherry picked from commit 76fc2fbc9801da5232cd4ba32d25298b74d460db)
---
pulsar-io/solr/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-io/solr/pom.xml b/pulsar-io/solr/pom.xml
index 7d7ceab..d6a5c76 100644
--- a/pulsar-io/solr/pom.xml
+++ b/pulsar-io/solr/pom.xml
@@ -29,7 +29,7 @@
</parent>
<properties>
- <solr.version>8.6.3</solr.version>
+ <solr.version>8.11.1</solr.version>
</properties>
<artifactId>pulsar-io-solr</artifactId>
[pulsar] 04/11: Getting rid of CVEs in batch-data-generator (#13820)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 444d2a3b49e803326c56bfaff7a461f13bd6ed4a
Author: Andrey Yegorov <86...@users.noreply.github.com>
AuthorDate: Wed Jan 19 23:51:57 2022 -0800
Getting rid of CVEs in batch-data-generator (#13820)
CVE-2021-41269
CVE-2021-22060
CVE-2021-22096
CVE-2021-22118
(cherry picked from commit 5a33d106dd282fa85008888a687c94203f888df7)
---
pom.xml | 4 ++--
pulsar-io/batch-data-generator/pom.xml | 6 ++++++
2 files changed, 8 insertions(+), 2 deletions(-)
diff --git a/pom.xml b/pom.xml
index 08b40df..1075d57 100644
--- a/pom.xml
+++ b/pom.xml
@@ -192,8 +192,8 @@ flexible messaging model and an intuitive client API.</description>
<jna.version>4.2.0</jna.version>
<kubernetesclient.version>12.0.1</kubernetesclient.version>
<nsq-client.version>1.0</nsq-client.version>
- <cron-utils.version>9.1.3</cron-utils.version>
- <spring-context.version>5.3.1</spring-context.version>
+ <cron-utils.version>9.1.6</cron-utils.version>
+ <spring-context.version>5.3.15</spring-context.version>
<apache-http-client.version>4.5.13</apache-http-client.version>
<!-- test dependencies -->
diff --git a/pulsar-io/batch-data-generator/pom.xml b/pulsar-io/batch-data-generator/pom.xml
index cdc1706..fe0b72e 100644
--- a/pulsar-io/batch-data-generator/pom.xml
+++ b/pulsar-io/batch-data-generator/pom.xml
@@ -45,6 +45,12 @@
</dependency>
<dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-context</artifactId>
+ <version>${spring-context.version}</version>
+ </dependency>
+
+ <dependency>
<groupId>io.codearte.jfairy</groupId>
<artifactId>jfairy</artifactId>
<version>0.5.9</version>
[pulsar] 01/11: Upgrade BK to 4.14.4 and Grpc to 1.42.1 (#13714)
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.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b85d06e5f9a692f4a57c2749e22978067477e5c4
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Tue Jan 11 14:47:28 2022 -0800
Upgrade BK to 4.14.4 and Grpc to 1.42.1 (#13714)
* Upgrade BK to 4.14.4
* Upgrade Grpc to 1.42.1
* Fixed license file
(cherry picked from commit 104adbe66ae327305227ce8f57ba3bb95ce83492)
---
distribution/server/src/assemble/LICENSE.bin.txt | 94 ++++++++++++------------
pom.xml | 4 +-
pulsar-sql/presto-distribution/LICENSE | 24 +++---
3 files changed, 61 insertions(+), 61 deletions(-)
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index e0458f2..c47458e 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -322,7 +322,7 @@ The Apache Software License, Version 2.0
- com.fasterxml.jackson.module-jackson-module-jsonSchema-2.12.6.jar
* Caffeine -- com.github.ben-manes.caffeine-caffeine-2.9.1.jar
* Conscrypt -- org.conscrypt-conscrypt-openjdk-uber-2.5.2.jar
- * Proto Google Common Protos -- com.google.api.grpc-proto-google-common-protos-1.17.0.jar
+ * Proto Google Common Protos -- com.google.api.grpc-proto-google-common-protos-2.0.1.jar
* Bitbucket -- org.bitbucket.b_c-jose4j-0.7.6.jar
* Gson
- com.google.code.gson-gson-2.8.9.jar
@@ -392,31 +392,31 @@ The Apache Software License, Version 2.0
- org.apache.logging.log4j-log4j-1.2-api-2.17.1.jar
* Java Native Access JNA -- net.java.dev.jna-jna-4.2.0.jar
* BookKeeper
- - org.apache.bookkeeper-bookkeeper-common-4.14.3.jar
- - org.apache.bookkeeper-bookkeeper-common-allocator-4.14.3.jar
- - org.apache.bookkeeper-bookkeeper-proto-4.14.3.jar
- - org.apache.bookkeeper-bookkeeper-server-4.14.3.jar
- - org.apache.bookkeeper-bookkeeper-tools-framework-4.14.3.jar
- - org.apache.bookkeeper-circe-checksum-4.14.3.jar
- - org.apache.bookkeeper-cpu-affinity-4.14.3.jar
- - org.apache.bookkeeper-statelib-4.14.3.jar
- - org.apache.bookkeeper-stream-storage-api-4.14.3.jar
- - org.apache.bookkeeper-stream-storage-common-4.14.3.jar
- - org.apache.bookkeeper-stream-storage-java-client-4.14.3.jar
- - org.apache.bookkeeper-stream-storage-java-client-base-4.14.3.jar
- - org.apache.bookkeeper-stream-storage-proto-4.14.3.jar
- - org.apache.bookkeeper-stream-storage-server-4.14.3.jar
- - org.apache.bookkeeper-stream-storage-service-api-4.14.3.jar
- - org.apache.bookkeeper-stream-storage-service-impl-4.14.3.jar
- - org.apache.bookkeeper.http-http-server-4.14.3.jar
- - org.apache.bookkeeper.http-vertx-http-server-4.14.3.jar
- - org.apache.bookkeeper.stats-bookkeeper-stats-api-4.14.3.jar
- - org.apache.bookkeeper.stats-prometheus-metrics-provider-4.14.3.jar
- - org.apache.distributedlog-distributedlog-common-4.14.3.jar
- - org.apache.distributedlog-distributedlog-core-4.14.3-tests.jar
- - org.apache.distributedlog-distributedlog-core-4.14.3.jar
- - org.apache.distributedlog-distributedlog-protocol-4.14.3.jar
- - org.apache.bookkeeper.stats-codahale-metrics-provider-4.14.3.jar
+ - org.apache.bookkeeper-bookkeeper-common-4.14.4.jar
+ - org.apache.bookkeeper-bookkeeper-common-allocator-4.14.4.jar
+ - org.apache.bookkeeper-bookkeeper-proto-4.14.4.jar
+ - org.apache.bookkeeper-bookkeeper-server-4.14.4.jar
+ - org.apache.bookkeeper-bookkeeper-tools-framework-4.14.4.jar
+ - org.apache.bookkeeper-circe-checksum-4.14.4.jar
+ - org.apache.bookkeeper-cpu-affinity-4.14.4.jar
+ - org.apache.bookkeeper-statelib-4.14.4.jar
+ - org.apache.bookkeeper-stream-storage-api-4.14.4.jar
+ - org.apache.bookkeeper-stream-storage-common-4.14.4.jar
+ - org.apache.bookkeeper-stream-storage-java-client-4.14.4.jar
+ - org.apache.bookkeeper-stream-storage-java-client-base-4.14.4.jar
+ - org.apache.bookkeeper-stream-storage-proto-4.14.4.jar
+ - org.apache.bookkeeper-stream-storage-server-4.14.4.jar
+ - org.apache.bookkeeper-stream-storage-service-api-4.14.4.jar
+ - org.apache.bookkeeper-stream-storage-service-impl-4.14.4.jar
+ - org.apache.bookkeeper.http-http-server-4.14.4.jar
+ - org.apache.bookkeeper.http-vertx-http-server-4.14.4.jar
+ - org.apache.bookkeeper.stats-bookkeeper-stats-api-4.14.4.jar
+ - org.apache.bookkeeper.stats-prometheus-metrics-provider-4.14.4.jar
+ - org.apache.distributedlog-distributedlog-common-4.14.4.jar
+ - org.apache.distributedlog-distributedlog-core-4.14.4-tests.jar
+ - org.apache.distributedlog-distributedlog-core-4.14.4.jar
+ - org.apache.distributedlog-distributedlog-protocol-4.14.4.jar
+ - org.apache.bookkeeper.stats-codahale-metrics-provider-4.14.4.jar
* Apache HTTP Client
- org.apache.httpcomponents-httpclient-4.5.13.jar
- org.apache.httpcomponents-httpcore-4.4.13.jar
@@ -456,21 +456,21 @@ The Apache Software License, Version 2.0
* Okio - com.squareup.okio-okio-1.13.0.jar
* Javassist -- org.javassist-javassist-3.25.0-GA.jar
* gRPC
- - io.grpc-grpc-all-1.33.0.jar
- - io.grpc-grpc-auth-1.33.0.jar
- - io.grpc-grpc-context-1.33.0.jar
- - io.grpc-grpc-core-1.33.0.jar
- - io.grpc-grpc-netty-1.33.0.jar
- - io.grpc-grpc-okhttp-1.33.0.jar
- - io.grpc-grpc-protobuf-1.33.0.jar
- - io.grpc-grpc-protobuf-lite-1.33.0.jar
- - io.grpc-grpc-stub-1.33.0.jar
- - io.grpc-grpc-alts-1.33.0.jar
- - io.grpc-grpc-api-1.33.0.jar
- - io.grpc-grpc-grpclb-1.33.0.jar
- - io.grpc-grpc-netty-shaded-1.33.0.jar
- - io.grpc-grpc-services-1.33.0.jar
- - io.grpc-grpc-xds-1.33.0.jar
+ - io.grpc-grpc-all-1.42.1.jar
+ - io.grpc-grpc-auth-1.42.1.jar
+ - io.grpc-grpc-context-1.42.1.jar
+ - io.grpc-grpc-core-1.42.1.jar
+ - io.grpc-grpc-netty-1.42.1.jar
+ - io.grpc-grpc-protobuf-1.42.1.jar
+ - io.grpc-grpc-protobuf-lite-1.42.1.jar
+ - io.grpc-grpc-stub-1.42.1.jar
+ - io.grpc-grpc-alts-1.42.1.jar
+ - io.grpc-grpc-api-1.42.1.jar
+ - io.grpc-grpc-grpclb-1.42.1.jar
+ - io.grpc-grpc-netty-shaded-1.42.1.jar
+ - io.grpc-grpc-services-1.42.1.jar
+ - io.grpc-grpc-xds-1.42.1.jar
+ - io.grpc-grpc-rls-1.42.1.jar
* Perfmark
- io.perfmark-perfmark-api-0.19.0.jar
* OpenCensus
@@ -517,15 +517,15 @@ The Apache Software License, Version 2.0
* Snappy Java
- org.xerial.snappy-snappy-java-1.1.7.jar
* Google HTTP Client
- - com.google.http-client-google-http-client-jackson2-1.34.0.jar
- - com.google.http-client-google-http-client-1.34.0.jar
- - com.google.auto.value-auto-value-annotations-1.7.jar
- - com.google.re2j-re2j-1.2.jar
+ - com.google.http-client-google-http-client-jackson2-1.38.0.jar
+ - com.google.http-client-google-http-client-1.38.0.jar
+ - com.google.auto.value-auto-value-annotations-1.7.4.jar
+ - com.google.re2j-re2j-1.5.jar
BSD 3-clause "New" or "Revised" License
* Google auth library
- - com.google.auth-google-auth-library-credentials-0.20.0.jar -- licenses/LICENSE-google-auth-library.txt
- - com.google.auth-google-auth-library-oauth2-http-0.20.0.jar -- licenses/LICENSE-google-auth-library.txt
+ - com.google.auth-google-auth-library-credentials-0.22.2.jar -- licenses/LICENSE-google-auth-library.txt
+ - com.google.auth-google-auth-library-oauth2-http-0.22.2.jar -- licenses/LICENSE-google-auth-library.txt
* LevelDB -- (included in org.rocksdb.*.jar) -- licenses/LICENSE-LevelDB.txt
* JSR305 -- com.google.code.findbugs-jsr305-3.0.2.jar -- licenses/LICENSE-JSR305.txt
diff --git a/pom.xml b/pom.xml
index c22850e..6bf2ce9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -103,7 +103,7 @@ flexible messaging model and an intuitive client API.</description>
<!-- apache commons -->
<commons-compress.version>1.21</commons-compress.version>
- <bookkeeper.version>4.14.3</bookkeeper.version>
+ <bookkeeper.version>4.14.4</bookkeeper.version>
<zookeeper.version>3.6.3</zookeeper.version>
<snappy.version>1.1.7</snappy.version> <!-- ZooKeeper server -->
<dropwizardmetrics.version>3.2.5</dropwizardmetrics.version> <!-- ZooKeeper server -->
@@ -131,7 +131,7 @@ flexible messaging model and an intuitive client API.</description>
<typetools.version>0.5.0</typetools.version>
<protobuf3.version>3.16.1</protobuf3.version>
<protoc3.version>${protobuf3.version}</protoc3.version>
- <grpc.version>1.33.0</grpc.version>
+ <grpc.version>1.42.1</grpc.version>
<perfmark.version>0.19.0</perfmark.version>
<protoc-gen-grpc-java.version>${grpc.version}</protoc-gen-grpc-java.version>
<gson.version>2.8.9</gson.version>
diff --git a/pulsar-sql/presto-distribution/LICENSE b/pulsar-sql/presto-distribution/LICENSE
index dbacae1..93499f4 100644
--- a/pulsar-sql/presto-distribution/LICENSE
+++ b/pulsar-sql/presto-distribution/LICENSE
@@ -414,18 +414,18 @@ The Apache Software License, Version 2.0
- async-http-client-2.12.1.jar
- async-http-client-netty-utils-2.12.1.jar
* Apache Bookkeeper
- - bookkeeper-common-4.14.3.jar
- - bookkeeper-common-allocator-4.14.3.jar
- - bookkeeper-proto-4.14.3.jar
- - bookkeeper-server-4.14.3.jar
- - bookkeeper-stats-api-4.14.3.jar
- - bookkeeper-tools-framework-4.14.3.jar
- - circe-checksum-4.14.3.jar
- - codahale-metrics-provider-4.14.3.jar
- - cpu-affinity-4.14.3.jar
- - http-server-4.14.3.jar
- - prometheus-metrics-provider-4.14.3.jar
- - codahale-metrics-provider-4.14.3.jar
+ - bookkeeper-common-4.14.4.jar
+ - bookkeeper-common-allocator-4.14.4.jar
+ - bookkeeper-proto-4.14.4.jar
+ - bookkeeper-server-4.14.4.jar
+ - bookkeeper-stats-api-4.14.4.jar
+ - bookkeeper-tools-framework-4.14.4.jar
+ - circe-checksum-4.14.4.jar
+ - codahale-metrics-provider-4.14.4.jar
+ - cpu-affinity-4.14.4.jar
+ - http-server-4.14.4.jar
+ - prometheus-metrics-provider-4.14.4.jar
+ - codahale-metrics-provider-4.14.4.jar
* Apache Commons
- commons-cli-1.2.jar
- commons-codec-1.15.jar