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/10 15:42:14 UTC
[pulsar] branch branch-2.8 updated (c75afb4 -> 0f5b699)
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 c75afb4 Fix time field use error (#12249)
new fc39643 Fix issues 11964, deadlock bug when use key_shared mode (#11965)
new ca38be4 Fix reader skipped remaining compacted data during the topic unloading. (#13629)
new a25c04e Update notice year. (#13653)
new 0f5b699 [Java Client] fixed Producer semaphore permit release issue (#13682)
The 4 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:
NOTICE | 2 +-
distribution/server/src/assemble/NOTICE.bin.txt | 2 +-
.../apache/bookkeeper/mledger/ManagedLedger.java | 3 +-
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 4 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 7 +--
.../mledger/impl/NonDurableCursorImpl.java | 23 ++++++---
.../AbstractDispatcherSingleActiveConsumer.java | 4 --
.../broker/service/persistent/PersistentTopic.java | 8 ++--
.../pulsar/client/impl/ProducerSemaphoreTest.java | 14 ++++++
.../pulsar/compaction/CompactedTopicTest.java | 55 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ProducerImpl.java | 16 +++++--
.../offload/jcloud/impl/MockManagedLedger.java | 4 +-
12 files changed, 114 insertions(+), 28 deletions(-)
[pulsar] 02/04: Fix reader skipped remaining compacted data during the topic unloading. (#13629)
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 ca38be4366435856aa77f089a813006e11d986fd
Author: lipenghui <pe...@apache.org>
AuthorDate: Fri Jan 7 13:44:02 2022 +0800
Fix reader skipped remaining compacted data during the topic unloading. (#13629)
To fix the reader skipping remaining compacted data while the topic has been unloaded.
with the earliest position. But if the reader has consumed some messages from the
compacted ledger but not all, the start position will not be `earliest`, the broker
will rewind the cursor for the reader to the next valid position of the original topic.
So the remaining messages in the compacted ledger will be skipped.
Here are the logs from the broker:
```
10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.pulsar.broker.service.BrokerService - Created topic persistent://xxx/product-full-prod/5126 - dedup is disabled
10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://xxx/product-full-prod/5126][xxx] Creating non-durable subscription at msg id 181759:14:-1:-1
10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.NonDurableCursorImpl - [xxx/product-full-prod/persistent/5126] Created non-durable cursor read-position=221199:0 mark-delete-position=181759:13
10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [xxx/product-full-prod/persistent/5126] Opened new cursor: NonDurableCursorImpl{ledger=xxx/product-full-prod/persistent/5126, ackPos=181759:13, readPos=221199:0}
10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [xxx/product-full-prod/persistent/5126-xxx] Rewind from 221199:0 to 221199:0
```
There some many compacted messages after `181759:13`, but the broker will not dispatch them to the reader.
The issue also can be reproduced by the unit test that was added in this PR.
If the cursor with `readCompacted = true`, just rewind to the next message of the mark delete position,
so that the reader can continue to read the data from the compacted ledger.
A new test added for testing the reader can get all the compacted messages and non-compacted messages from the topic during the topic unloading.
(cherry picked from commit 07f131fec7db36d6f424ce419d26888592b04207)
---
.../apache/bookkeeper/mledger/ManagedLedger.java | 3 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 7 +--
.../mledger/impl/NonDurableCursorImpl.java | 23 ++++++---
.../AbstractDispatcherSingleActiveConsumer.java | 4 --
.../broker/service/persistent/PersistentTopic.java | 8 ++--
.../pulsar/compaction/CompactedTopicTest.java | 55 ++++++++++++++++++++++
.../offload/jcloud/impl/MockManagedLedger.java | 4 +-
7 files changed, 84 insertions(+), 20 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
index 74c67e9..0200e25 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
@@ -259,7 +259,8 @@ public interface ManagedLedger {
*/
ManagedCursor newNonDurableCursor(Position startCursorPosition) throws ManagedLedgerException;
ManagedCursor newNonDurableCursor(Position startPosition, String subscriptionName) throws ManagedLedgerException;
- ManagedCursor newNonDurableCursor(Position startPosition, String subscriptionName, InitialPosition initialPosition) throws ManagedLedgerException;
+ ManagedCursor newNonDurableCursor(Position startPosition, String subscriptionName, InitialPosition initialPosition,
+ boolean isReadCompacted) throws ManagedLedgerException;
/**
* Delete a ManagedCursor asynchronously.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 4fe5e9a..9ee0a8f 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -1030,11 +1030,12 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
@Override
public ManagedCursor newNonDurableCursor(Position startPosition, String subscriptionName) throws ManagedLedgerException {
- return newNonDurableCursor(startPosition, subscriptionName, InitialPosition.Latest);
+ return newNonDurableCursor(startPosition, subscriptionName, InitialPosition.Latest, false);
}
@Override
- public ManagedCursor newNonDurableCursor(Position startCursorPosition, String cursorName, InitialPosition initialPosition)
+ public ManagedCursor newNonDurableCursor(Position startCursorPosition, String cursorName, InitialPosition initialPosition,
+ boolean isReadCompacted)
throws ManagedLedgerException {
Objects.requireNonNull(cursorName, "cursor name can't be null");
checkManagedLedgerIsOpen();
@@ -1049,7 +1050,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
NonDurableCursorImpl cursor = new NonDurableCursorImpl(bookKeeper, config, this, cursorName,
- (PositionImpl) startCursorPosition, initialPosition);
+ (PositionImpl) startCursorPosition, initialPosition, isReadCompacted);
cursor.setActive();
log.info("[{}] Opened new cursor: {}", name, cursor);
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
index 0f7ffe41..1d545bd 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java
@@ -33,11 +33,12 @@ import org.slf4j.LoggerFactory;
public class NonDurableCursorImpl extends ManagedCursorImpl {
- private volatile boolean readCompacted;
+ private final boolean readCompacted;
NonDurableCursorImpl(BookKeeper bookkeeper, ManagedLedgerConfig config, ManagedLedgerImpl ledger, String cursorName,
- PositionImpl startCursorPosition, CommandSubscribe.InitialPosition initialPosition) {
+ PositionImpl startCursorPosition, CommandSubscribe.InitialPosition initialPosition, boolean isReadCompacted) {
super(bookkeeper, config, ledger, cursorName);
+ this.readCompacted = isReadCompacted;
// Compare with "latest" position marker by using only the ledger id. Since the C++ client is using 48bits to
// store the entryId, it's not able to pass a Long.max() as entryId. In this case there's no point to require
@@ -67,7 +68,7 @@ public class NonDurableCursorImpl extends ManagedCursorImpl {
private void recoverCursor(PositionImpl mdPosition) {
Pair<PositionImpl, Long> lastEntryAndCounter = ledger.getLastPositionAndCounter();
- this.readPosition = ledger.getNextValidPosition(mdPosition);
+ this.readPosition = isReadCompacted() ? mdPosition.getNext() : ledger.getNextValidPosition(mdPosition);
markDeletePosition = mdPosition;
// Initialize the counter such that the difference between the messages written on the ML and the
@@ -118,15 +119,23 @@ public class NonDurableCursorImpl extends ManagedCursorImpl {
callback.deleteCursorComplete(ctx);
}
- public void setReadCompacted(boolean readCompacted) {
- this.readCompacted = readCompacted;
- }
-
public boolean isReadCompacted() {
return readCompacted;
}
@Override
+ public void rewind() {
+ // For reading the compacted data,
+ // we couldn't reset the read position to the next valid position of the original topic.
+ // Otherwise, the remaining data in the compacted ledger will be skipped.
+ if (!readCompacted) {
+ super.rewind();
+ } else {
+ readPosition = markDeletePosition.getNext();
+ }
+ }
+
+ @Override
public synchronized String toString() {
return MoreObjects.toStringHelper(this).add("ledger", ledger.getName()).add("ackPos", markDeletePosition)
.add("readPos", readPosition).toString();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
index 4c7ea45..8cab06b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
@@ -27,7 +27,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import org.apache.bookkeeper.mledger.ManagedCursor;
-import org.apache.bookkeeper.mledger.impl.NonDurableCursorImpl;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
@@ -181,9 +180,6 @@ public abstract class AbstractDispatcherSingleActiveConsumer extends AbstractBas
consumer.notifyActiveConsumerChange(currentActiveConsumer);
}
}
- if (cursor != null && !cursor.isDurable() && cursor instanceof NonDurableCursorImpl) {
- ((NonDurableCursorImpl) cursor).setReadCompacted(ACTIVE_CONSUMER_UPDATER.get(this).readCompacted());
- }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 6ad10fb..87728a8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -750,7 +750,7 @@ public class PersistentTopic extends AbstractTopic
getDurableSubscription(subscriptionName, initialPosition, startMessageRollbackDurationSec,
replicatedSubscriptionState)
: getNonDurableSubscription(subscriptionName, startMessageId, initialPosition,
- startMessageRollbackDurationSec);
+ startMessageRollbackDurationSec, readCompacted);
int maxUnackedMessages = isDurable
? getMaxUnackedMessagesOnConsumer()
@@ -895,7 +895,8 @@ public class PersistentTopic extends AbstractTopic
}
private CompletableFuture<? extends Subscription> getNonDurableSubscription(String subscriptionName,
- MessageId startMessageId, InitialPosition initialPosition, long startMessageRollbackDurationSec) {
+ MessageId startMessageId, InitialPosition initialPosition, long startMessageRollbackDurationSec,
+ boolean isReadCompacted) {
log.info("[{}][{}] Creating non-durable subscription at msg id {}", topic, subscriptionName, startMessageId);
CompletableFuture<Subscription> subscriptionFuture = new CompletableFuture<>();
@@ -928,7 +929,8 @@ public class PersistentTopic extends AbstractTopic
Position startPosition = new PositionImpl(ledgerId, entryId);
ManagedCursor cursor = null;
try {
- cursor = ledger.newNonDurableCursor(startPosition, subscriptionName, initialPosition);
+ cursor = ledger.newNonDurableCursor(startPosition, subscriptionName, initialPosition,
+ isReadCompacted);
} catch (ManagedLedgerException e) {
return FutureUtil.failedFuture(e);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
index 4d00d28..accce4ef 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
@@ -734,4 +734,59 @@ public class CompactedTopicTest extends MockedPulsarServiceBaseTest {
Assert.assertNull(reader.readNext(3, TimeUnit.SECONDS));
}
+ public void testReadCompleteMessagesDuringTopicUnloading() throws Exception {
+ String topic = "persistent://my-property/use/my-ns/testReadCompleteMessagesDuringTopicUnloading-" +
+ UUID.randomUUID();
+ final int numMessages = 1000;
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic)
+ .blockIfQueueFull(true)
+ .enableBatching(false)
+ .create();
+ CompletableFuture<MessageId> lastMessage = null;
+ for (int i = 0; i < numMessages; ++i) {
+ lastMessage = producer.newMessage().key(i + "").value(String.format("msg [%d]", i)).sendAsync();
+ }
+ producer.flush();
+ lastMessage.join();
+ admin.topics().triggerCompaction(topic);
+ Awaitility.await().untilAsserted(() -> {
+ PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic);
+ Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1);
+ Assert.assertEquals(stats.compactedLedger.entries, numMessages);
+ Assert.assertEquals(admin.topics().getStats(topic)
+ .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0);
+ Assert.assertEquals(stats.lastConfirmedEntry, stats.cursors.get(COMPACTION_SUBSCRIPTION).markDeletePosition);
+ });
+ // Unload the topic to make sure the original ledger been deleted.
+ admin.topics().unload(topic);
+ // Produce more messages to the original topic
+ for (int i = 0; i < numMessages; ++i) {
+ lastMessage = producer.newMessage().key(i + numMessages + "").value(String.format("msg [%d]", i + numMessages)).sendAsync();
+ }
+ producer.flush();
+ lastMessage.join();
+ // For now the topic has 1000 messages in the compacted ledger and 1000 messages in the original topic.
+ @Cleanup
+ Reader<String> reader = pulsarClient.newReader(Schema.STRING)
+ .topic(topic)
+ .startMessageIdInclusive()
+ .startMessageId(MessageId.earliest)
+ .readCompacted(true)
+ .create();
+
+ // Unloading the topic during reading the data to make sure the reader will not miss any messages.
+ for (int i = 0; i < numMessages / 2; ++i) {
+ Assert.assertEquals(reader.readNext().getValue(), String.format("msg [%d]", i));
+ }
+ admin.topics().unload(topic);
+ for (int i = 0; i < numMessages / 2; ++i) {
+ Assert.assertEquals(reader.readNext().getValue(), String.format("msg [%d]", i + numMessages / 2));
+ }
+ admin.topics().unload(topic);
+ for (int i = 0; i < numMessages; ++i) {
+ Assert.assertEquals(reader.readNext().getValue(), String.format("msg [%d]", i + numMessages));
+ }
+ }
}
diff --git a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java
index 767190c..229cd66 100644
--- a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java
+++ b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java
@@ -122,8 +122,8 @@ public class MockManagedLedger implements ManagedLedger {
@Override
public ManagedCursor newNonDurableCursor(Position startPosition, String subscriptionName,
- CommandSubscribe.InitialPosition initialPosition) throws
- ManagedLedgerException {
+ CommandSubscribe.InitialPosition initialPosition,
+ boolean isReadCompacted) throws ManagedLedgerException {
return null;
}
[pulsar] 04/04: [Java Client] fixed Producer semaphore permit release issue (#13682)
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 0f5b6993a4fccd789e6a42278e8d1d7f0b4528cf
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Mon Jan 10 19:27:26 2022 +0800
[Java Client] fixed Producer semaphore permit release issue (#13682)
(cherry picked from commit 916b61d3a497c44c8511347d2c4eac95c53f8506)
---
.../apache/pulsar/client/impl/ProducerSemaphoreTest.java | 14 ++++++++++++++
.../java/org/apache/pulsar/client/impl/ProducerImpl.java | 16 +++++++++++-----
2 files changed, 25 insertions(+), 5 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
index f46509e..d325ed6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
@@ -75,11 +75,13 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
futures.add(producer.newMessage().value(("Semaphore-test-" + i).getBytes()).sendAsync());
}
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize - messages);
+ Assert.assertFalse(producer.isErrorStat());
} finally {
producer.getClientCnx().channel().config().setAutoRead(true);
}
FutureUtil.waitForAll(futures).get();
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
futures.clear();
// Simulate replicator, non batching message but `numMessagesInBatch` of message metadata > 1
@@ -92,15 +94,18 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
futures.add(producer.sendAsync(msg));
}
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize - messages/2);
+ Assert.assertFalse(producer.isErrorStat());
} finally {
producer.getClientCnx().channel().config().setAutoRead(true);
}
FutureUtil.waitForAll(futures).get();
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
futures.clear();
// Here must ensure that the semaphore available permits is 0
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
// Acquire 5 and not wait the send ack call back
producer.getClientCnx().channel().config().setAutoRead(false);
@@ -111,12 +116,14 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
// Here must ensure that the Semaphore a acquired 5
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize - messages / 2);
+ Assert.assertFalse(producer.isErrorStat());
} finally {
producer.getClientCnx().channel().config().setAutoRead(true);
}
FutureUtil.waitForAll(futures).get();
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
}
/**
@@ -141,6 +148,7 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
// Test that when we fill the queue with "replicator" messages, we are notified
// (replicator itself would block)
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
producer.getClientCnx().channel().config().setAutoRead(false);
try {
for (int i = 0; i < pendingQueueSize; i++) {
@@ -151,6 +159,7 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
futures.add(producer.sendAsync(msg));
}
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), 0);
+ Assert.assertFalse(producer.isErrorStat());
try {
MessageMetadata metadata = new MessageMetadata()
.setNumMessagesInBatch(10);
@@ -162,6 +171,7 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
Assert.assertEquals(ee.getCause().getClass(),
PulsarClientException.ProducerQueueIsFullError.class);
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), 0);
+ Assert.assertFalse(producer.isErrorStat());
}
} finally {
producer.getClientCnx().channel().config().setAutoRead(true);
@@ -171,12 +181,14 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
// Test that when we fill the queue with normal messages, we get an error
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
producer.getClientCnx().channel().config().setAutoRead(false);
try {
for (int i = 0; i < pendingQueueSize; i++) {
futures.add(producer.newMessage().value(("Semaphore-test-" + i).getBytes()).sendAsync());
}
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), 0);
+ Assert.assertFalse(producer.isErrorStat());
try {
producer.newMessage().value(("Semaphore-test-Q-full").getBytes()).sendAsync().get();
@@ -184,6 +196,7 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
Assert.assertEquals(ee.getCause().getClass(),
PulsarClientException.ProducerQueueIsFullError.class);
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), 0);
+ Assert.assertFalse(producer.isErrorStat());
}
} finally {
@@ -191,5 +204,6 @@ public class ProducerSemaphoreTest extends ProducerConsumerBase {
}
FutureUtil.waitForAll(futures).get();
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), pendingQueueSize);
+ Assert.assertFalse(producer.isErrorStat());
}
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index 78aaf3c..cb1bc47 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -256,11 +256,12 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
protected void semaphoreRelease(final int releaseCountRequest) {
if (semaphore.isPresent()) {
if (!errorState) {
- final int availablePermits = semaphore.get().availablePermits();
- if (availablePermits - releaseCountRequest < 0) {
- log.error("Semaphore permit release count request greater then availablePermits" +
- " : availablePermits={}, releaseCountRequest={}",
- availablePermits, releaseCountRequest);
+ final int availableReleasePermits =
+ conf.getMaxPendingMessages() - this.semaphore.get().availablePermits();
+ if (availableReleasePermits - releaseCountRequest < 0) {
+ log.error("Semaphore permit release count request greater then availableReleasePermits" +
+ " : availableReleasePermits={}, releaseCountRequest={}",
+ availableReleasePermits, releaseCountRequest);
errorState = true;
}
}
@@ -2000,5 +2001,10 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
return semaphore;
}
+ @VisibleForTesting
+ boolean isErrorStat() {
+ return errorState;
+ }
+
private static final Logger log = LoggerFactory.getLogger(ProducerImpl.class);
}
\ No newline at end of file
[pulsar] 01/04: Fix issues 11964, deadlock bug when use key_shared mode (#11965)
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 fc396430805b47caadd95f1365a74d29e1861ef2
Author: baomingyu <ba...@163.com>
AuthorDate: Fri Jan 7 12:51:41 2022 +0800
Fix issues 11964, deadlock bug when use key_shared mode (#11965)
(cherry picked from commit fd3ba55f8e39992ecbc96830af5ee58e3247a25a)
---
.../java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index f22b327..61a5aa3 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -1992,7 +1992,6 @@ public class ManagedCursorImpl implements ManagedCursor {
if (individualDeletedMessages.isEmpty()) {
// No changes to individually deleted messages, so nothing to do at this point
- callback.deleteComplete(ctx);
return;
}
@@ -2024,6 +2023,9 @@ public class ManagedCursorImpl implements ManagedCursor {
return;
} finally {
lock.writeLock().unlock();
+ if (individualDeletedMessages.isEmpty()) {
+ callback.deleteComplete(ctx);
+ }
}
// Apply rate limiting to mark-delete operations
[pulsar] 03/04: Update notice year. (#13653)
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 a25c04e49510020727cb634dcb2a71ddbbea838a
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Sat Jan 8 03:30:07 2022 +0800
Update notice year. (#13653)
### Motivation
Update NOTICE from 2021-2022
(cherry picked from commit bacac04f564536be9c30481d2cab88512906967e)
---
NOTICE | 2 +-
distribution/server/src/assemble/NOTICE.bin.txt | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/NOTICE b/NOTICE
index 2052417..bbbe4fa 100644
--- a/NOTICE
+++ b/NOTICE
@@ -1,5 +1,5 @@
Apache Pulsar
-Copyright 2017-2021 The Apache Software Foundation
+Copyright 2017-2022 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
diff --git a/distribution/server/src/assemble/NOTICE.bin.txt b/distribution/server/src/assemble/NOTICE.bin.txt
index 661921c..bc5e2e6 100644
--- a/distribution/server/src/assemble/NOTICE.bin.txt
+++ b/distribution/server/src/assemble/NOTICE.bin.txt
@@ -1,6 +1,6 @@
Apache Pulsar
-Copyright 2017-2021 The Apache Software Foundation
+Copyright 2017-2022 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).