You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/03/20 01:52:57 UTC
[pulsar] branch branch-2.11 updated: [fix][client] moving get sequenceId into the sync code segment (#17836) (#19837)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.11 by this push:
new aeac08d9258 [fix][client] moving get sequenceId into the sync code segment (#17836) (#19837)
aeac08d9258 is described below
commit aeac08d925859f057a81d9910b43c57a8eb1cf39
Author: Xiangying Meng <55...@users.noreply.github.com>
AuthorDate: Mon Mar 20 09:52:50 2023 +0800
[fix][client] moving get sequenceId into the sync code segment (#17836) (#19837)
---
.../pulsar/client/api/ClientDeduplicationTest.java | 50 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ProducerImpl.java | 29 ++++++-------
2 files changed, 64 insertions(+), 15 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java
index c8acc7d46f8..0ac0440a7a8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java
@@ -28,6 +28,9 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
@@ -373,4 +376,51 @@ public class ClientDeduplicationTest extends ProducerConsumerBase {
consumer.close();
producer.close();
}
+
+ @Test
+ public void testUpdateSequenceIdInSyncCodeSegment() throws Exception {
+ final String topic = "persistent://my-property/my-ns/testUpdateSequenceIdInSyncCodeSegment";
+ int totalMessage = 200;
+ int threadSize = 5;
+ String topicName = "subscription";
+ ExecutorService executorService = Executors.newFixedThreadPool(threadSize);
+ conf.setBrokerDeduplicationEnabled(true);
+
+ //build producer/consumer
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topic)
+ .producerName("producer")
+ .sendTimeout(0, TimeUnit.SECONDS)
+ .create();
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionName(topicName)
+ .subscribe();
+
+ CountDownLatch countDownLatch = new CountDownLatch(threadSize);
+ //Send messages in multiple-thread
+ for (int i = 0; i < threadSize; i++) {
+ executorService.submit(() -> {
+ try {
+ for (int j = 0; j < totalMessage; j++) {
+ //The message will be sent with out-of-order sequence ID.
+ producer.newMessage().sendAsync();
+ }
+ } catch (Exception e) {
+ log.error("Failed to send/ack messages with transaction.", e);
+ } finally {
+ countDownLatch.countDown();
+ }
+ });
+ }
+ //wait the all send op is executed and store its futures in the arraylist.
+ countDownLatch.await();
+
+ for (int i = 0; i < threadSize * totalMessage; i++) {
+ Message<byte[]> msg = consumer.receive(5, TimeUnit.SECONDS);
+ assertNotNull(msg);
+ }
+ }
}
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 6f7d7e6a148..3969d1b2afe 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
@@ -101,7 +101,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// Producer id, used to identify a producer within a single connection
protected final long producerId;
- // Variable is used through the atomic updater
+ // Variable is updated in a synchronized block
private volatile long msgIdGenerator;
private final OpSendMsgQueue pendingMessages;
@@ -169,10 +169,6 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
private boolean errorState;
- @SuppressWarnings("rawtypes")
- private static final AtomicLongFieldUpdater<ProducerImpl> msgIdGeneratorUpdater = AtomicLongFieldUpdater
- .newUpdater(ProducerImpl.class, "msgIdGenerator");
-
public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfigurationData conf,
CompletableFuture<Producer<T>> producerCreatedFuture, int partitionIndex, Schema<T> schema,
ProducerInterceptors interceptors, Optional<String> overrideProducerName) {
@@ -489,7 +485,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// Update the message metadata before computing the payload chunk size to avoid a large message cannot be split
// into chunks.
- final long sequenceId = updateMessageMetadata(msgMetadata, uncompressedSize);
+ updateMessageMetadata(msgMetadata, uncompressedSize);
// send in chunks
int totalChunks;
@@ -529,6 +525,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
try {
synchronized (this) {
int readStartIndex = 0;
+ final long sequenceId = updateMessageMetadataSequenceId(msgMetadata);
String uuid = totalChunks > 1 ? String.format("%s-%d", producerName, sequenceId) : null;
ChunkedMessageCtx chunkedMessageCtx = totalChunks > 1 ? ChunkedMessageCtx.get(totalChunks) : null;
byte[] schemaVersion = totalChunks > 1 && msg.getMessageBuilder().hasSchemaVersion()
@@ -570,15 +567,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
* @param uncompressedSize
* @return the sequence id
*/
- private long updateMessageMetadata(final MessageMetadata msgMetadata, final int uncompressedSize) {
- final long sequenceId;
- if (!msgMetadata.hasSequenceId()) {
- sequenceId = msgIdGeneratorUpdater.getAndIncrement(this);
- msgMetadata.setSequenceId(sequenceId);
- } else {
- sequenceId = msgMetadata.getSequenceId();
- }
-
+ private void updateMessageMetadata(final MessageMetadata msgMetadata, final int uncompressedSize) {
if (!msgMetadata.hasPublishTime()) {
msgMetadata.setPublishTime(client.getClientClock().millis());
@@ -592,6 +581,16 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
}
msgMetadata.setUncompressedSize(uncompressedSize);
}
+ }
+
+ private long updateMessageMetadataSequenceId(final MessageMetadata msgMetadata) {
+ final long sequenceId;
+ if (!msgMetadata.hasSequenceId()) {
+ sequenceId = msgIdGenerator++;
+ msgMetadata.setSequenceId(sequenceId);
+ } else {
+ sequenceId = msgMetadata.getSequenceId();
+ }
return sequenceId;
}