From dc8ca9750022a6af9b03e764ed142628c8ffa417 Mon Sep 17 00:00:00 2001 From: liangyepianzhou Date: Mon, 26 Sep 2022 16:37:38 +0800 Subject: [PATCH 01/11] [Fix][client] Make get sequenceId in the sync code segment ### Motivation When the producer send message in the multiple thread, the message with the smaller sequenceId can be push later than the message with the bigger sequenceId. Then there will be exception thrown at persistentTopic::publishTxnMessage ### Modification Move getting sequenceId in the sync code. --- .../java/org/apache/pulsar/client/impl/ProducerImpl.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) 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 c51562157df5f..d83fbf5fd2e70 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 @@ -487,10 +487,6 @@ public void sendAsync(Message message, SendCallback callback) { return; } - // 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); - // send in chunks int totalChunks; int payloadChunkSize; @@ -528,6 +524,10 @@ public void sendAsync(Message message, SendCallback callback) { try { 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 = updateMessageMetadata(msgMetadata, uncompressedSize); + int readStartIndex = 0; String uuid = totalChunks > 1 ? String.format("%s-%d", producerName, sequenceId) : null; ChunkedMessageCtx chunkedMessageCtx = totalChunks > 1 ? ChunkedMessageCtx.get(totalChunks) : null; From 69f675524e63753804ff57e1c914bcdd0185ae61 Mon Sep 17 00:00:00 2001 From: liangyepianzhou Date: Mon, 26 Sep 2022 22:17:27 +0800 Subject: [PATCH 02/11] add test --- .../broker/transaction/TransactionTest.java | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index ffc351e941353..d03a4cd4acd20 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -52,6 +52,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -1469,4 +1470,63 @@ public void testTBRecoverChangeStateError() throws InterruptedException, Timeout Assert.assertTrue(t instanceof BrokerServiceException.ServiceUnitNotReadyException); } } + + @Test + public void testUpdateSequenceIdInSyncCodeSegment() throws Exception { + String topic = NAMESPACE1 + "/sequenceId"; + int totalMessage = 10; + int threadSize = 30; + String topicName = "subscription"; + ExecutorService executorService = Executors.newFixedThreadPool(threadSize); + + //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(); + + //send and ack messages with transaction + Transaction transaction = pulsarClient.newTransaction() + .withTransactionTimeout(10, TimeUnit.SECONDS) + .build() + .get(); + + for (int i = 0; i < totalMessage * threadSize; i++) { + producer.newMessage().send(); + } + + CountDownLatch countDownLatch = new CountDownLatch(threadSize); + new Thread(() -> { + for (int i = 0; i < threadSize; i++) { + executorService.submit(() -> { + try { + for (int j = 0; j < totalMessage; j++) { + producer.newMessage(transaction).sendAsync(); + Message message = consumer.receive(); + consumer.acknowledgeAsync(message.getMessageId(), + transaction); + } + countDownLatch.countDown(); + } catch (Exception e) { + log.error("Failed to send/ack messages with transaction.", e); + } + }); + } + }).start(); + //wait the all send/ack op is excuted and store its futures in the arraylist. + countDownLatch.await(10, TimeUnit.SECONDS); + transaction.commit().get(); + + for (int i = 0; i < threadSize * totalMessage; i++) { + consumer.receive(); + } + + } } From 729f5e5b86ff52f73226ae0a57036cdaee289ff8 Mon Sep 17 00:00:00 2001 From: liangyepianzhou Date: Fri, 7 Oct 2022 23:42:49 +0800 Subject: [PATCH 03/11] optimize test --- .../broker/transaction/TransactionTest.java | 38 ++++++++----------- 1 file changed, 16 insertions(+), 22 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index 404b7ff0259c8..3e06e5e0d756c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -1551,30 +1551,24 @@ public void testUpdateSequenceIdInSyncCodeSegment() throws Exception { } CountDownLatch countDownLatch = new CountDownLatch(threadSize); - new Thread(() -> { - for (int i = 0; i < threadSize; i++) { - executorService.submit(() -> { - try { - for (int j = 0; j < totalMessage; j++) { - producer.newMessage(transaction).sendAsync(); - Message message = consumer.receive(); - consumer.acknowledgeAsync(message.getMessageId(), - transaction); - } - countDownLatch.countDown(); - } catch (Exception e) { - log.error("Failed to send/ack messages with transaction.", e); + for (int i = 0; i < threadSize; i++) { + executorService.submit(() -> { + try { + for (int j = 0; j < totalMessage; j++) { + producer.newMessage(transaction).sendAsync(); + Message message = consumer.receive(); + consumer.acknowledgeAsync(message.getMessageId(), + transaction); } - }); - } - }).start(); - //wait the all send/ack op is excuted and store its futures in the arraylist. + } catch (Exception e) { + log.error("Failed to send/ack messages with transaction.", e); + } finally { + countDownLatch.countDown(); + } + }); + } + //wait the all send/ack op is executed and store its futures in the arraylist. countDownLatch.await(10, TimeUnit.SECONDS); transaction.commit().get(); - - for (int i = 0; i < threadSize * totalMessage; i++) { - consumer.receive(); - } - } } From f317d5d7839c28730de49e3bbd4d16fb8af52f39 Mon Sep 17 00:00:00 2001 From: liangyepianzhou Date: Mon, 10 Oct 2022 10:47:45 +0800 Subject: [PATCH 04/11] optimize --- .../org/apache/pulsar/broker/transaction/TransactionTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index 3e06e5e0d756c..8de78c03a2ed9 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -1542,7 +1542,7 @@ public void testUpdateSequenceIdInSyncCodeSegment() throws Exception { //send and ack messages with transaction Transaction transaction = pulsarClient.newTransaction() - .withTransactionTimeout(10, TimeUnit.SECONDS) + .withTransactionTimeout(5, TimeUnit.SECONDS) .build() .get(); @@ -1568,7 +1568,7 @@ public void testUpdateSequenceIdInSyncCodeSegment() throws Exception { }); } //wait the all send/ack op is executed and store its futures in the arraylist. - countDownLatch.await(10, TimeUnit.SECONDS); + countDownLatch.await(5, TimeUnit.SECONDS); transaction.commit().get(); } } From db2b0f53af121da976df623da8e56cd3e7ee3a7d Mon Sep 17 00:00:00 2001 From: liangyepianzhou Date: Mon, 10 Oct 2022 11:10:15 +0800 Subject: [PATCH 05/11] optimize --- .../org/apache/pulsar/broker/transaction/TransactionTest.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index 8de78c03a2ed9..ed7910c0b462e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -1555,6 +1555,7 @@ public void testUpdateSequenceIdInSyncCodeSegment() throws Exception { executorService.submit(() -> { try { for (int j = 0; j < totalMessage; j++) { + //The message will be sent with out-of-order sequence ID. producer.newMessage(transaction).sendAsync(); Message message = consumer.receive(); consumer.acknowledgeAsync(message.getMessageId(), @@ -1569,6 +1570,7 @@ public void testUpdateSequenceIdInSyncCodeSegment() throws Exception { } //wait the all send/ack op is executed and store its futures in the arraylist. countDownLatch.await(5, TimeUnit.SECONDS); + //The transaction will be failed due to timeout. transaction.commit().get(); } } From 320608cbd31f8cba22d3a755ff30a01f7e129a5d Mon Sep 17 00:00:00 2001 From: liangyepianzhou Date: Wed, 23 Nov 2022 20:38:24 +0800 Subject: [PATCH 06/11] move the test to ClientDeduplicationTest --- .../broker/transaction/TransactionTest.java | 54 ------------------- .../client/api/ClientDeduplicationTest.java | 50 +++++++++++++++++ 2 files changed, 50 insertions(+), 54 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index 40d41928a3c42..0e818653abbae 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -1616,58 +1616,4 @@ public void testGetTxnState() throws Exception { Awaitility.await().until(() -> abortingTxn.getState() == Transaction.State.ABORTING); } - @Test - public void testUpdateSequenceIdInSyncCodeSegment() throws Exception { - String topic = NAMESPACE1 + "/sequenceId"; - int totalMessage = 10; - int threadSize = 30; - String topicName = "subscription"; - ExecutorService executorService = Executors.newFixedThreadPool(threadSize); - - //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(); - - //send and ack messages with transaction - Transaction transaction = pulsarClient.newTransaction() - .withTransactionTimeout(5, TimeUnit.SECONDS) - .build() - .get(); - - for (int i = 0; i < totalMessage * threadSize; i++) { - producer.newMessage().send(); - } - - CountDownLatch countDownLatch = new CountDownLatch(threadSize); - 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(transaction).sendAsync(); - Message message = consumer.receive(); - consumer.acknowledgeAsync(message.getMessageId(), - transaction); - } - } catch (Exception e) { - log.error("Failed to send/ack messages with transaction.", e); - } finally { - countDownLatch.countDown(); - } - }); - } - //wait the all send/ack op is executed and store its futures in the arraylist. - countDownLatch.await(5, TimeUnit.SECONDS); - //The transaction will be failed due to timeout. - transaction.commit().get(); - } } 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..4797602bb67e5 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 testUpdateSequenceIdInSyncCodeSe7gment() throws Exception { + final String topic = "persistent://my-property/my-ns/testUpdateSequenceIdInSyncCodeSegment"; + int totalMessage = 10; + int threadSize = 30; + 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); + } + } } From b68c4f9b1e6c8daeac6ff702323abb8e1043bb10 Mon Sep 17 00:00:00 2001 From: liangyepianzhou Date: Wed, 23 Nov 2022 20:52:27 +0800 Subject: [PATCH 07/11] move the test to ClientDeduplicationTest --- .../org/apache/pulsar/broker/transaction/TransactionTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java index 0e818653abbae..33305a2b8df10 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TransactionTest.java @@ -1615,5 +1615,4 @@ public void testGetTxnState() throws Exception { Transaction abortingTxn = transaction; Awaitility.await().until(() -> abortingTxn.getState() == Transaction.State.ABORTING); } - } From f9b09b217479c9ba9cacf15853869c05e2334d60 Mon Sep 17 00:00:00 2001 From: liangyepianzhou Date: Thu, 24 Nov 2022 18:50:55 +0800 Subject: [PATCH 08/11] fix some comments --- .../apache/pulsar/client/api/ClientDeduplicationTest.java | 6 +++--- 1 file changed, 3 insertions(+), 3 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 4797602bb67e5..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 @@ -377,10 +377,10 @@ public void testKeyBasedBatchingOrder() throws Exception { } @Test - public void testUpdateSequenceIdInSyncCodeSe7gment() throws Exception { + public void testUpdateSequenceIdInSyncCodeSegment() throws Exception { final String topic = "persistent://my-property/my-ns/testUpdateSequenceIdInSyncCodeSegment"; - int totalMessage = 10; - int threadSize = 30; + int totalMessage = 200; + int threadSize = 5; String topicName = "subscription"; ExecutorService executorService = Executors.newFixedThreadPool(threadSize); conf.setBrokerDeduplicationEnabled(true); From 411d1211314687f2ac8671622c19a022085d8e1a Mon Sep 17 00:00:00 2001 From: liangyepianzhou Date: Thu, 24 Nov 2022 19:03:05 +0800 Subject: [PATCH 09/11] LineLength --- .../main/java/org/apache/pulsar/client/impl/ProducerImpl.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 2c6b2130c57d3..bf4f328e62261 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 @@ -549,8 +549,8 @@ 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. + // 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); String uuid = totalChunks > 1 ? String.format("%s-%d", producerName, sequenceId) : null; From ae58e4ba5c608e10d6c17f2ba711380aed768ad3 Mon Sep 17 00:00:00 2001 From: liangyepianzhou Date: Thu, 24 Nov 2022 22:23:55 +0800 Subject: [PATCH 10/11] fix some comments --- .../java/org/apache/pulsar/client/impl/ProducerImpl.java | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) 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 bf4f328e62261..a3379c6c6b1f9 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) { @@ -579,7 +575,7 @@ public void sendAsync(Message message, SendCallback callback) { private long updateMessageMetadata(final MessageMetadata msgMetadata, final int uncompressedSize) { final long sequenceId; if (!msgMetadata.hasSequenceId()) { - sequenceId = msgIdGeneratorUpdater.getAndIncrement(this); + sequenceId = msgIdGenerator++; msgMetadata.setSequenceId(sequenceId); } else { sequenceId = msgMetadata.getSequenceId(); From 460a38abf12eaba1fed028b00dc83d7476c6acb9 Mon Sep 17 00:00:00 2001 From: xiangying <1984997880@qq.com> Date: Tue, 13 Dec 2022 09:25:11 +0800 Subject: [PATCH 11/11] fix --- .../pulsar/client/impl/ProducerImpl.java | 25 +++++++++++-------- 1 file changed, 15 insertions(+), 10 deletions(-) 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 e506cd5a4f340..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 @@ -481,6 +481,9 @@ public void sendAsync(Message message, SendCallback callback) { return; } + // Update the message metadata before computing the payload chunk size to avoid a large message cannot be split + // into chunks. + updateMessageMetadata(msgMetadata, uncompressedSize); // send in chunks int totalChunks; @@ -549,7 +552,7 @@ public void sendAsync(Message message, SendCallback callback) { 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 = updateMessageMetadata(msgMetadata, uncompressedSize); + final long sequenceId = updateMessageMetadataSequenceId(msgMetadata); String uuid = totalChunks > 1 ? String.format("%s-%d", producerName, sequenceId) : null; serializeAndSendMessage(msg, payload, sequenceId, uuid, chunkId, totalChunks, @@ -574,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 = msgIdGenerator++; - msgMetadata.setSequenceId(sequenceId); - } else { - sequenceId = msgMetadata.getSequenceId(); - } - + private void updateMessageMetadata(final MessageMetadata msgMetadata, final int uncompressedSize) { if (!msgMetadata.hasPublishTime()) { msgMetadata.setPublishTime(client.getClientClock().millis()); @@ -596,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; }