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 168886f733a3c..d2f9617a5faae 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 @@ -27,6 +27,9 @@ 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; @@ -372,4 +375,51 @@ public void testKeyBasedBatchingOrder() throws Exception { 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 producer = pulsarClient.newProducer() + .topic(topic) + .producerName("producer") + .sendTimeout(0, TimeUnit.SECONDS) + .create(); + + Consumer 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 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 dceaf50214182..ef3ead790bc61 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 extends ProducerBase 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 extends ProducerBase implements TimerTask, Conne private boolean errorState; - @SuppressWarnings("rawtypes") - private static final AtomicLongFieldUpdater msgIdGeneratorUpdater = AtomicLongFieldUpdater - .newUpdater(ProducerImpl.class, "msgIdGenerator"); - public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfigurationData conf, CompletableFuture> producerCreatedFuture, int partitionIndex, Schema schema, ProducerInterceptors interceptors, Optional overrideProducerName) { @@ -487,7 +483,7 @@ public void sendAsync(Message message, SendCallback callback) { // 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; @@ -527,7 +523,6 @@ public void sendAsync(Message message, SendCallback callback) { try { int readStartIndex = 0; - 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() ? msg.getMessageBuilder().getSchemaVersion() : null; @@ -555,6 +550,11 @@ public void sendAsync(Message message, SendCallback callback) { return; } synchronized (this) { + // Update the message metadata before computing the payload chunk size + // to avoid a large message cannot be split into chunks. + final long sequenceId = updateMessageMetadataSequenceId(msgMetadata); + String uuid = totalChunks > 1 ? String.format("%s-%d", producerName, sequenceId) : null; + serializeAndSendMessage(msg, payload, sequenceId, uuid, chunkId, totalChunks, readStartIndex, payloadChunkSize, compressedPayload, compressed, compressedPayload.readableBytes(), callback, chunkedMessageCtx, messageId); @@ -577,15 +577,7 @@ public void sendAsync(Message message, SendCallback callback) { * @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()); @@ -599,6 +591,16 @@ private long updateMessageMetadata(final MessageMetadata msgMetadata, final int } 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; }