[Issue 5476]Fix message deduplicate issue while using external sequence id with batch produce #5491

codelipenghui · 2019-10-29T09:23:27Z

Fixes #5476

Motivation

Fix #5476

Modifications

Add last_sequence_id in MessageMetadata and CommandSend, use sequence id and last_sequence_id to indicate the batch lowest_sequence_id and highest_sequence_id.
Handle batch message deduplicate check in MessageDeduplication
Response the last_sequence_id to client and add message deduplicate check in client

Verifying this change

Added new unit tests to verify this change

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (yes)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

Documentation

Does this pull request introduce a new feature? (no)

codelipenghui · 2019-10-29T09:34:34Z

@sijie

It's hard to handle sequence id in key based batcher, if user set an external sequence id, we can't rewrite it and the sequence ids will distributed into inner batchers of key based batcher.

Current implementation will throw exception if client can ensure the message is duplicated, since we do not throws exceptions before, is it ok to throws exception?

codelipenghui · 2019-10-30T11:37:46Z

run java8 tests

wolfstudy · 2019-10-30T13:35:32Z

hello @codelipenghui

can you fix the test case:

org.apache.pulsar.broker.service.ReplicatorTest.testResumptionAfterBacklogRelaxed

codelipenghui · 2019-10-30T16:33:05Z

run java8 tests

merlimat · 2019-10-30T22:51:19Z

pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationFailureTest.java

+        try {
+            producer.newMessage().sequenceId(producerThread.getLastSeqId() + 1).value("end").send();
+            fail("should failed, because send a duplication");
+        } catch (PulsarClientException.InvalidMessageException ignore) {


I don't think this is a valid assumption. The logic of dedup is to transparently perform the deduplication, without giving error to the application in case of duplicates.

merlimat · 2019-10-30T22:58:51Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

+                    if (sequenceId <= lastSequenceIdPublished) {
+                        callback.sendComplete(new PulsarClientException
+                                .InvalidMessageException("Message is definitely a duplicate"));


We shouldn't trigger an error to the application when there are dups, rather the contract is that it gets an OK. In any case we need to think through the implication of triggering an error just for one message out of band.
This is a very different behavior from the other failure modes where all the messages are failed after 1 failure.

I believe that handling this in broker side is much preferable.

If users use the external sequence id and enable batch on producer, sequence id 1,2,3,1 will happens if we do not the check, we can just throw an exception when using the external sequence id?

merlimat · 2019-10-30T23:04:00Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

+                        callback.sendComplete(new PulsarClientException
+                                .InvalidMessageException("Message is a definitely a duplicate or not cannot be " +
+                                "determined at this time"));


If we don't know whether is dup or not, then why are we triggering error here?

merlimat · 2019-10-30T23:06:33Z

pulsar-common/src/main/proto/PulsarApi.proto

+
+    /// Add lowest and highest sequence id to support external sequence id
+    optional uint64 lowest_sequence_id = 6 [default = 0];
+    optional uint64 highest_sequence_id = 7 [default = 0];


Why do we need 2 new fields here? Don't we just need to a new last_sequence_id? isn't lowest_sequence_id the same as sequence_id ?

Yes, we can only add a new last_sequence_id, will fix.

merlimat · 2019-10-30T23:08:12Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java

@@ -76,6 +76,14 @@ default long getOriginalSequenceId() {
        }

        void completed(Exception e, long ledgerId, long entryId);
+
+        default long getLowestSequenceId() {


I don't see how these sequence ids relate to a Topic.

What's the lowest sequenceId ? I think this only apply to 1 single batch.

Sequence id are per producer anyway

It is in the inner class named PublishContext of Topic

It is in the inner class named PublishContext of Topic

Which already contains a sequence id field

merlimat · 2019-10-30T23:09:11Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java

@@ -257,6 +285,9 @@ public ServerCnx getCnx() {
        private String originalProducerName;
        private long originalSequenceId;

+        private long lowestSequenceId;


What does it mean the lowest sequence id of a producer?

It is in the inner class named MessagePublishContext of Producer

Ok, though as in other comment, there's already a sequenceId field. Is that now ignored? If yes, then it should be removed.

Also, as you can see, there's an originalSequenceId field. This is used in the context of geo-replication and it would have to be accounted for as well.

codelipenghui · 2019-10-31T03:36:46Z

@merlimat Thanks for the review, i have addressed your comments, please take a look again.

merlimat · 2019-10-31T17:11:30Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java

@@ -257,6 +285,9 @@ public ServerCnx getCnx() {
        private String originalProducerName;
        private long originalSequenceId;

+        private long lowestSequenceId;


Ok, though as in other comment, there's already a sequenceId field. Is that now ignored? If yes, then it should be removed.

Also, as you can see, there's an originalSequenceId field. This is used in the context of geo-replication and it would have to be accounted for as well.

merlimat · 2019-10-31T17:12:14Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java

@@ -76,6 +76,14 @@ default long getOriginalSequenceId() {
        }

        void completed(Exception e, long ledgerId, long entryId);
+
+        default long getLowestSequenceId() {


It is in the inner class named PublishContext of Topic

Which already contains a sequence id field

merlimat · 2019-10-31T17:16:07Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -367,6 +371,11 @@ public void sendAsync(Message<T> message, SendCallback callback) {
                    msgMetadataBuilder.setSequenceId(sequenceId);
                } else {
                    sequenceId = msgMetadataBuilder.getSequenceId();
+                    if (sequenceId <= lastSequenceIdPushed) {
+                        callback.sendComplete(new PulsarClientException
+                                .InvalidMessageException("Message is definitely a duplicate"));


This is not correct on 2 levels:

If sequenceId <= lastSequenceIdPushed we don't know yet whether the message is already dup, because the previous attempt might still fail. This has to be disambiguated by the broker which has visibility at the storage level.

As mentioned before, we cannot throw error when there's a duplicate, rather we need to return "ok" to the application.

codelipenghui · 2019-11-01T14:46:27Z

org.apache.pulsar.broker.service.PeerReplicatorTest.testPeerClusterInReplicationClusterListChang

run java8 tests
run integration tests

codelipenghui · 2019-11-02T05:35:34Z

run java8 tests
run integration tests

codelipenghui · 2019-11-02T05:36:03Z

@merlimat Please help take a look, i have addressed your comments

codelipenghui · 2019-11-03T01:33:20Z

run java8 tests

sijie · 2019-11-05T03:10:49Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java

@@ -298,7 +344,8 @@ public void completed(Exception exception, long ledgerId, long entryId) {
                    if (!(exception instanceof TopicClosedException)) {
                        // For TopicClosed exception there's no need to send explicit error, since the client was
                        // already notified
-                        producer.cnx.ctx().writeAndFlush(Commands.newSendError(producer.producerId, sequenceId,
+                        long callBackSequenceId = lastSequenceId >= sequenceId ? lastSequenceId : sequenceId;


can this be simplified with the following statement?

long callbackSequenceId = Math.max(lastSequenceId, sequenceId);

I will add a common function to get the callback sequenceId

sijie · 2019-11-05T03:11:00Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java

@@ -330,8 +377,9 @@ public void run() {
            // stats
            rateIn.recordMultipleEvents(batchSize, msgSize);
            producer.topic.recordAddLatency(System.nanoTime() - startTimeNs, TimeUnit.NANOSECONDS);
+            long callBackSequenceId = lastSequenceId >= sequenceId ? lastSequenceId : sequenceId;;


same comment as above

sijie · 2019-11-05T03:12:43Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java

@@ -80,6 +81,12 @@ public void add(MessageImpl<?> msg, SendCallback callback) {
        previousCallback = callback;
        currentBatchSizeBytes += msg.getDataBuffer().readableBytes();
        messages.add(msg);
+        if (lowestSequenceId == -1) {


nit:

Suggested change

if (lowestSequenceId == -1) {

if (lowestSequenceId == -1L) {

sijie · 2019-11-05T03:12:58Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java

@@ -50,7 +50,8 @@

    private PulsarApi.MessageMetadata.Builder messageMetadata = PulsarApi.MessageMetadata.newBuilder();
    // sequence id for this batch which will be persisted as a single entry by broker
-    private long sequenceId = -1;
+    private long lowestSequenceId = -1;


Suggested change

private long lowestSequenceId = -1;

private long lowestSequenceId = -1L;

sijie · 2019-11-05T03:13:06Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java

@@ -50,7 +50,8 @@

    private PulsarApi.MessageMetadata.Builder messageMetadata = PulsarApi.MessageMetadata.newBuilder();
    // sequence id for this batch which will be persisted as a single entry by broker
-    private long sequenceId = -1;
+    private long lowestSequenceId = -1;
+    private long highestSequenceId = -1;


Suggested change

private long highestSequenceId = -1;

private long highestSequenceId = -1L;

sijie · 2019-11-05T03:18:22Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -776,8 +798,7 @@ void ackReceived(ClientCnx cnx, long sequenceId, long ledgerId, long entryId) {
                }
                return;
            }
-
-            long expectedSequenceId = op.sequenceId;
+            long expectedSequenceId = op.highestSequenceId > 0 ? op.highestSequenceId : op.sequenceId;


can you make a common function for this statement here?

sijie · 2019-11-05T03:18:52Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -803,7 +824,7 @@ void ackReceived(ClientCnx cnx, long sequenceId, long ledgerId, long entryId) {
        if (callback) {
            op = pendingCallbacks.poll();
            if (op != null) {
-                lastSequenceIdPublished = op.sequenceId + op.numMessagesInBatch - 1;
+                lastSequenceIdPublished = sequenceId;


should this be highest sequence id?

sijie · 2019-11-05T03:20:40Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -1391,6 +1429,9 @@ private void processOpSendMsg(OpSendMsg op) {
                batchMessageAndSend();
            }
            pendingMessages.put(op);
+            if (op.msg != null) {
+                lastSequenceIdPushed = op.sequenceId;


I think we need to keep track highestSequenceId, no?

We should only assign the highest sequence id, no?

lastSequenceIdPushed = Math.max(lastSequenceIdPushed, op.lastSequenceIdPushed);

sijie · 2019-11-05T03:21:47Z

pulsar-common/src/main/proto/PulsarApi.proto

@@ -127,6 +127,9 @@ message MessageMetadata {
    // transaction related message info
    optional uint64 txnid_least_bits = 22 [default = 0];
    optional uint64 txnid_most_bits = 23 [default = 0];
+
+    /// Add last sequence id to support batch message with external sequence id
+    optional uint64 last_sequence_id = 24 [default = 0];


It seems to me that we use highest sequence id at the client side, but use last at the wire protocol and broker side. Can we make it consistent by just using highest across the places?

sijie · 2019-11-05T03:22:56Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java

+
+    public void publishMessage(long producerId, long sequenceId, long lastSequenceId,
+           ByteBuf headersAndPayload, long batchSize) {
+        if (sequenceId > lastSequenceId) {


sequenceId => lowestSequenceId
lastSequenceId => highestSequenceId

It is very confusing using sequenceId and lastSequenceId.

codelipenghui · 2019-11-05T14:43:15Z

run java8 tests
run cpp tests

codelipenghui · 2019-11-05T16:24:38Z

run java8 tests

codelipenghui · 2019-11-06T00:40:01Z

run java8 tests

wolfstudy · 2019-11-06T01:59:25Z

ping @sijie @merlimat PTAL again.

run java8 tests

codelipenghui · 2019-11-06T02:39:37Z

run java8 tests

sijie · 2019-11-06T06:18:27Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -803,7 +824,7 @@ void ackReceived(ClientCnx cnx, long sequenceId, long ledgerId, long entryId) {
        if (callback) {
            op = pendingCallbacks.poll();
            if (op != null) {
-                lastSequenceIdPublished = op.sequenceId + op.numMessagesInBatch - 1;
+                lastSequenceIdPublished = getHighestSequenceId(op);


this should :

lastSequenceIdPublished = Math.max(lastSequenceIdPublished, getHighestSequenceId(op));

because the duplicated publishes will still succeed, right. It can override a larger sequence Id with a smaller one.

sijie · 2019-11-06T06:20:03Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -839,7 +864,7 @@ protected synchronized void recoverChecksumError(ClientCnx cnx, long sequenceId)
                log.debug("[{}] [{}] Got send failure for timed out msg {}", topic, producerName, sequenceId);
            }
        } else {
-            long expectedSequenceId = op.sequenceId;
+            long expectedSequenceId = op.highestSequenceId > 0 ? op.highestSequenceId : op.sequenceId;


expectedSequenceId = getHighestSequenceId(op);

codelipenghui · 2019-11-06T08:14:55Z

run java8 tests

codelipenghui · 2019-11-06T08:58:32Z

run java8 tests

codelipenghui · 2019-11-06T12:55:54Z

run java8 tests

codelipenghui · 2019-11-08T09:02:00Z

run java8 tests

codelipenghui · 2019-11-08T10:24:21Z

run java8 tests

codelipenghui · 2019-11-09T01:17:36Z

run java8 tests

codelipenghui · 2019-11-09T01:18:13Z

run cpp tests

wolfstudy · 2019-11-09T05:49:59Z

run java8 tests

wolfstudy · 2019-11-09T05:50:20Z

ping @merlimat PTAL again.

codelipenghui · 2019-11-10T00:42:46Z

run java8 tests

codelipenghui · 2019-11-10T06:49:27Z

run java8 tests

codelipenghui · 2019-11-10T14:15:17Z

run java8 tests

codelipenghui · 2019-11-11T02:00:31Z

ping @merlimat PTAL again.

wolfstudy · 2019-11-12T02:07:48Z

@merlimat PTAL again, thanks

merlimat

👍 Nice work @codelipenghui

wolfstudy · 2019-11-14T02:19:04Z

@codelipenghui The change is not compatible with 2.4.2, i will move the Milestone to 2.5.0.

## Motivation Since #5491 merged, while user use new pulsar client to produce batch messages to older version broker(e.g. 2.4.0), send ack error will occur: ``` [pulsar-client-io-8-2] WARN org.apache.pulsar.client.impl.ProducerImpl - [persistent://sandbox/pressure-test/test-A-partition-11] [pulsar-cluster-test-13-294] Got ack for msg. expecting: 13 - got: 224 - queue-size: 9 ``` The problem is client use highest sequence id to match the response sequence id, but in old version broker can not return the highest id. So, this pr is try to fix the problem of produce batch message with new version client and old version broker. ### Modifications Add highest sequence id to CommandSendReceipt. If the response highest sequence id of send receipt > lowest sequence id, it means broker is a new version broker, so we need to verify the highest sequence id, otherwise we only verify the lowest sequence id.

codelipenghui requested review from sijie, jerrypeng, merlimat, jiazhai and wolfstudy October 29, 2019 09:23

codelipenghui self-assigned this Oct 29, 2019

codelipenghui added this to the 2.4.2 milestone Oct 29, 2019

merlimat requested changes Oct 30, 2019

View reviewed changes

merlimat reviewed Oct 31, 2019

View reviewed changes

sijie requested changes Nov 5, 2019

View reviewed changes

codelipenghui requested review from merlimat and sijie November 6, 2019 02:05

sijie requested changes Nov 6, 2019

View reviewed changes

codelipenghui added 8 commits November 8, 2019 15:15

Use only one last sequence id.

f1c1eb2

fix comments

126522c

fix comments

b4d158d

Fix comments

5554017

Fix comments

ba5e91c

Fix params name

b0375ee

Fix comments

cf2f4da

End with L for long type fields

d982c8d

codelipenghui force-pushed the issue-5476 branch from a29257d to d982c8d Compare November 8, 2019 07:16

Merge branch 'master' into issue-5476

02658af

merlimat approved these changes Nov 13, 2019

View reviewed changes

sijie merged commit d3cb108 into apache:master Nov 13, 2019

codelipenghui mentioned this pull request Nov 13, 2019

Add highest sequenceId for CommandSendReceipt. #5653

Merged

wolfstudy modified the milestones: 2.4.2, 2.5.0 Nov 14, 2019

codelipenghui deleted the issue-5476 branch May 19, 2021 05:37

BewareMyPower mentioned this pull request May 2, 2022

[Java Client] Fix wrong behavior of deduplication for key based batching #15413

Closed

4 tasks

	private long lowestSequenceId = -1;
	private long lowestSequenceId = -1L;

	private long highestSequenceId = -1;
	private long highestSequenceId = -1L;

[Issue 5476]Fix message deduplicate issue while using external sequence id with batch produce #5491

[Issue 5476]Fix message deduplicate issue while using external sequence id with batch produce #5491

Conversation

codelipenghui commented Oct 29, 2019 • edited Loading

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

codelipenghui commented Oct 29, 2019

codelipenghui commented Oct 30, 2019

wolfstudy commented Oct 30, 2019

codelipenghui commented Oct 30, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Oct 31, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Nov 1, 2019

codelipenghui commented Nov 2, 2019

codelipenghui commented Nov 2, 2019

codelipenghui commented Nov 3, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Nov 5, 2019

codelipenghui commented Nov 5, 2019

codelipenghui commented Nov 6, 2019

wolfstudy commented Nov 6, 2019

codelipenghui commented Nov 6, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Nov 6, 2019

codelipenghui commented Nov 6, 2019

codelipenghui commented Nov 6, 2019

codelipenghui commented Nov 8, 2019

codelipenghui commented Nov 8, 2019

codelipenghui commented Nov 9, 2019

codelipenghui commented Nov 9, 2019

wolfstudy commented Nov 9, 2019

wolfstudy commented Nov 9, 2019

codelipenghui commented Nov 10, 2019

codelipenghui commented Nov 10, 2019

codelipenghui commented Nov 10, 2019

codelipenghui commented Nov 11, 2019

wolfstudy commented Nov 12, 2019

merlimat left a comment

Choose a reason for hiding this comment

wolfstudy commented Nov 14, 2019

codelipenghui commented Oct 29, 2019 •

edited

Loading