You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2023/04/18 13:49:28 UTC
[GitHub] [pulsar] poorbarcode opened a new pull request, #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
poorbarcode opened a new pull request, #20128:
URL: https://github.com/apache/pulsar/pull/20128
### Motivation
**Background of Deduplication:** Every message will have an attribute `sequence-id`, and Broker will mark the last `sequence-id` for each producer. If a producer sends a message with a smaller `sequence-id`, it will be rejected.
---
**Background of Geo-Replication:** The replicator works like this: read messages from the source cluster; send messages to the target cluster; if there has something error, Replicator will rewind these messages; loop to the next reading.
**Background of ManagedCursor:** ManagedCursor marks the next position which will read from BK, and we call it `readPosition`, and update after a read is complete. For example:
1. initialize `readPosition` with `0.`
2. read 10 messages
3. update `readPosition` to `10.`
---
<strong>(Highlight)</strong>Therefore, if messages are sent out of order, many messages will be discarded. for example: if send `1,2,3,5,4`, then the message `4` will be discarded by duplication check.
---
**There have two scenarios that make messages out of order**
- scenario-1
- read messages from the original cluster
- read messages from the original cluster
- receive messages `[1,2,3]`(the callback of the first read)
- send messages `[1,2,3]` to the remote cluster, but something is wrong. Replicator trigger `rewind` and try to send again.
- receive messages `[4,5]`(the callback of the second read)
- send messages `[4,5]` to remote cluster.
- receive messages `[1,2,3]` after `rewind`.
- scenario-2
| time | `thread rewind` | `thread reading messages`|
| -- | --- | --- |
| 1 | receive messages `[3:0 ~ 4:0]` | |
| 1 | | async read start(`readPosition=4:0, markDeleted=3:0`) |
| 2 | rewind `readPosition` (` --> 3:0`)` | |
| 3 | | read complete, set `readPosition`(` --> 5:0`) |
| 4 | receive messages `[4:0 ~ 5.0]` | |
| 5 | the messages `3:0 ~ 4:0` will nolanger be received until next `rewind` |
### Modifications
Make the replicator processes messages only sequentially
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository:
- https://github.com/poorbarcode/pulsar/pull/87
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171247540
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java:
##########
@@ -268,6 +275,36 @@ protected void readMoreEntries() {
}
}
+ /**
+ * Determine if the messages are continuous, if they are not, the messages are discarded and rewind is called.
+ * Why are the messages discontinuous, for example: if there has something wrong when Replicator is processing
+ * messages "[1:1 ~ 3:3]", Replicator discards the unprocessed message. But a new batch messages "[4:1 ~ 6:6]"
+ * is received later, then these messages will be sent.
+ */
+ protected boolean isMessageContinuousAndRewindIfNot(List<Entry> entries) {
+ if (CollectionUtils.isEmpty(entries)){
+ return true;
+ }
+
+ PositionImpl markDeletedPos = (PositionImpl) cursor.getMarkDeletedPosition();
+ PositionImpl lastSentPos = (PositionImpl) lastSent;
+ PositionImpl lastProcessedPos = markDeletedPos.compareTo(lastSentPos) > 0 ? markDeletedPos : lastSentPos;
+
+ PositionImpl firstMessageReceived = (PositionImpl) entries.get(0).getPosition();
+ PositionImpl expectedFirstMessage = managedLedger.getNextValidPosition(lastProcessedPos);
+ if (expectedFirstMessage.compareTo(firstMessageReceived) >= 0) {
+ return true;
+ }
Review Comment:
To make the logic of `replicateEntries` easier to read, I changed the method name to `checkNoMessageSkipped`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171147881
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
Will the current solution make the replicator not able to re-replicate data(reset the cursor of the replicator) to the remote cluster unless restart the replicator?
And will the solution be resolved by unloading the topic if the replicator runs into an exception?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #20128: [draft] [fix] [broker] fix messages lost in scenario geo-replication if enabled deduplication
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#issuecomment-1705833568
The pr had no activity for 30 days, mark with Stale label.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171183018
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
@codelipenghui
> Will the current solution make the replicator not able to re-replicate data(reset the cursor of the replicator) to the remote cluster unless restart the replicator?
1. If someone moves Replicator's cursor to a larger pointer manually(such as `pulsar-admin topics resetCursor`), everything is ok. The logic of checking "whether the message is continuous or not" is `first_message_received <= Max(mark_deleted_pos, last_sent_pos) + 1`, since `resetCursor` will move the variable `markDeletePosition` to a larger value too, the continuous check will return `true`.
2. If someone moves Replicator's cursor to a smaller pointer manually(such as `rewind` or `pulsar-admin topics resetCursor`), everything is ok.
> And will the solution be resolved by unloading the topic if the replicator runs into an exception?
If something makes the variable `lastSent` to a wrong value, it can be solved by `unload topic`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on pull request #20128: [draft] [fix] [broker] fix messages lost in scenario geo-replication if enabled deduplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#issuecomment-1666071900
This PR will continue after waiting for [PIP-269](https://github.com/apache/pulsar/pull/20469) to complete.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1170768645
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
Hi @codelipenghui
> What kind of exceptions will happen to the producer? IMO, we should use infinite publish timeout, if the exception is not publish timeout, it should not be a retriable exception. We should stop the replication.
My description of the scene wasn't good, it is not an `exception` for sending messages. Now there are two scenarios:
- the schema of messages is loading, and Replicator will call `rewind` this scenario. See Code-1 below
- something is wrong when processing `replicateEntries`, and Replicator will not call `rewind` this scenario. See Code-2 below
#### Code-1
https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java#L160-L179
#### Code-2
https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java#L190-L192
I opened a comment list to easily track the context(^_^).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1170784504
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
@codelipenghui
> Ok, if we can reach https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java#L190-L192, can we just close the replicator and try to restart the replication task?
Background: Now the Replicator is having some trouble starting and closing:
- After closing the internal producer of Replicator, this Replicator will not work anymore, there has no mechanism to get it running again
- I will push another PR to fix it.
- The function `disconnect` maybe has a bug that makes the Replicator continues to work after the internal cursor is closed
- When `startProducer` and `disconnect` run concurrently, the thread `disconnect` may not get the right producer.
- Maybe something is wrong else. I'm looking into it
- I will push another PR to fix it.
So there is not yet a stable way to restart Replicator(other than an `unload topic`).
---
### Summary
Now this PR tries to solve three problems in the same way: `Make the replicator processes messages only sequentially`
- scenario-1: race condition of `read entries` and `rewind`
- scenario-2: the task replicate messages is aborted by a loading schema
- scenario-3: the task replicate messages is aborted by an unknown error(we are talking about)
Since scenario-1 and scenario-2 are there, the solution `Make the replicator processes messages only sequentially` is needed, so we can do no additional processing on scenario-3.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled deduplication
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#issuecomment-1556062728
The pr had no activity for 30 days, mark with Stale label.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171214746
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
> I mean if we don't apply this fix but try to "close the replicator -> unload the topic" if the replicator reaches https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java#L190-L192, is it able to fix the issue?
Yes, I think so.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171358564
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
Ok, this way looks easier to understand and straightforward.
It's very similar to Pulsar connectors, like a connector to consume messages from a topic and publish them to another topic. I mean if want to introduce a principle to handle such issues, I think we don't want to add such logic to check if there are messages been skipped. This is why I think we should not introduce such logic here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1170784504
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
@codelipenghui
> Ok, if we can reach https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java#L190-L192, can we just close the replicator and try to restart the replication task?
Background: Now the Replicator is having some trouble starting and closing:
- After closing the internal producer of Replicator, this Replicator will not work anymore, there has no mechanism to get it running again
- I will push another PR to fix it.
- The function `disconnect` maybe has a bug that makes the Replicator continues to work after the internal cursor is closed
- When `startProducer` and `disconnect` run concurrently, the thread `disconnect` may not get the right producer.
- I will push another PR to fix it.
So there is not yet a stable way to restart Replicator(other than an `unload topic`).
---
### Summary
Now this PR tries to solve three problems in the same way: `Make the replicator processes messages only sequentially`
- scenario-1: race condition of `read entries` and `rewind`
- scenario-2: the task replicate messages is aborted by a loading schema
- scenario-3: the task replicate messages is aborted by an unknown error(we are talking about)
Since scenario-1 and scenario-2 are there, the solution `Make the replicator processes messages only sequentially` is needed, so we can do no additional processing on scenario-3.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171435654
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
@codelipenghui I think this suggestion is good. Thanks. I will try to solve the issues described above with multiple PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1170768645
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
Hi @codelipenghui
> What kind of exceptions will happen to the producer? IMO, we should use infinite publish timeout, if the exception is not publish timeout, it should not be a retriable exception. We should stop the replication.
My description of the scene wasn't good, it is not an `exception` for sending messages. Now there are two scenarios:
- the schema of messages is loading. see Code-1 below
- something wrong when processing `replicateEntries`. see Code-2 below
#### Code-1
https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java#L160-L179
#### Code-2
https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java#L190-L192
I opened a comment list to easily track the context(^_^).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171183018
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
@codelipenghui
> Will the current solution make the replicator not able to re-replicate data(reset the cursor of the replicator) to the remote cluster unless restart the replicator?
1. If someone moves Replicator's cursor to a larger pointer manually(such as `pulsar-admin topics resetCursor`), everything is ok. The logic of checking "whether the message is continuous or not" is `first_message_received <= Max(mark_deleted_pos, last_sent_pos) + 1`, since `resetCursor` will move the variable `markDeletePosition` to a larger value too, the continuous check will return `true`.
2. If someone moves Replicator's cursor to a smaller pointer manually(such as `rewind` or `pulsar-admin topics resetCursor`), everything is ok.
> And will the solution be resolved by unloading the topic if the replicator runs into an exception?
I think so.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171183018
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
@codelipenghui
> Will the current solution make the replicator not able to re-replicate data(reset the cursor of the replicator) to the remote cluster unless restart the replicator?
1. If someone moves Replicator's cursor to a larger pointer manually(such as `pulsar-admin topics resetCursor`), everything is ok. The logic of checking "whether the message is continuous or not" is `first_message_received <= Max(mark_deleted_pos, last_sent_pos) + 1`, since `resetCursor` will move the variable `markDeletePosition` to a larger value too, the continuous check will return `true`.
2. If someone moves Replicator's cursor to a smaller pointer manually(such as `rewind`), everything is ok.
> And will the solution be resolved by unloading the topic if the replicator runs into an exception?
I think so.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1170771638
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
Ok, if we can reach https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java#L190-L192, can we just close the replicator and try to restart the replication task?
The main purpose of the description of the PR is to avoid processing [4,5] before [1,2,3] is processed successfully. It should be a corner case with a very low incidence. If close the issue replicator can help with the problem, it will be more simpler for implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171233767
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java:
##########
@@ -268,6 +275,36 @@ protected void readMoreEntries() {
}
}
+ /**
+ * Determine if the messages are continuous, if they are not, the messages are discarded and rewind is called.
+ * Why are the messages discontinuous, for example: if there has something wrong when Replicator is processing
+ * messages "[1:1 ~ 3:3]", Replicator discards the unprocessed message. But a new batch messages "[4:1 ~ 6:6]"
+ * is received later, then these messages will be sent.
+ */
+ protected boolean isMessageContinuousAndRewindIfNot(List<Entry> entries) {
+ if (CollectionUtils.isEmpty(entries)){
+ return true;
+ }
+
+ PositionImpl markDeletedPos = (PositionImpl) cursor.getMarkDeletedPosition();
+ PositionImpl lastSentPos = (PositionImpl) lastSent;
+ PositionImpl lastProcessedPos = markDeletedPos.compareTo(lastSentPos) > 0 ? markDeletedPos : lastSentPos;
+
+ PositionImpl firstMessageReceived = (PositionImpl) entries.get(0).getPosition();
+ PositionImpl expectedFirstMessage = managedLedger.getNextValidPosition(lastProcessedPos);
+ if (expectedFirstMessage.compareTo(firstMessageReceived) >= 0) {
+ return true;
+ }
Review Comment:
Good suggestion. Already fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171251110
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java:
##########
@@ -113,6 +117,7 @@ protected boolean replicateEntries(List<Entry> entries) {
// Increment pending messages for messages produced locally
PENDING_MESSAGES_UPDATER.incrementAndGet(this);
+ lastSent = entry.getPosition();
Review Comment:
The reply for this comment is same as https://github.com/apache/pulsar/pull/20128#discussion_r1171246436
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171246436
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java:
##########
@@ -183,6 +188,7 @@ protected boolean replicateEntries(List<Entry> entries) {
msg.getMessageBuilder().clearTxnidLeastBits();
// Increment pending messages for messages produced locally
PENDING_MESSAGES_UPDATER.incrementAndGet(this);
+ lastSent = entry.getPosition();
Review Comment:
> It looks like the inflight messages will also be treated as lastSent?
Yes, it was done on purpose.
---
Because there has a scenario like this:
- start sending messages `1~3` to the remote cluster, and it is not finished yet(`lastSent` is 0 now).
- <strong>(Highlight)</strong>receive messages `4~5`. these messages will be discarded and trigger a rewind. This will reduce the efficiency of task Replication.
---
If we update the variable `lastSent` when the sending is started(this PR does), it will works like this:
- start sending messages `1~3` to the remote cluster, and it is not finished yet(`lastSent` is 3 now).
- receive messages `4~5` and send them(`lastSent` is 5 now).
- <strong>(Highlight)</strong>the message `3` failed to send, call `rewind`(`markDeleted` is 2 now) and set the variable `lastSent` to `earliest`.
- Because the timeout of the producer is `0`, the probability of this scenario occurring is very, very small.
- receive the message `3` the second time. because `3 > max(2, earliest)`, the check `hasMessageSkipped` will be pass
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171234880
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java:
##########
@@ -113,6 +117,7 @@ protected boolean replicateEntries(List<Entry> entries) {
// Increment pending messages for messages produced locally
PENDING_MESSAGES_UPDATER.incrementAndGet(this);
+ lastSent = entry.getPosition();
Review Comment:
add the context: the linked comment is https://github.com/apache/pulsar/pull/20128/files#r1171223753
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171423751
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
> This scenario is hard to fix. Because there is no clear marker of which in-flight reads need to be discarded after step-2. I think I should learn how the connector handle messages.
I see. We can continue to handle these messages after loading schema is finished.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171211658
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
> If something makes the variable lastSent to a wrong value, it can be solved by unload topic
@poorbarcode I mean if we don't apply this fix but try to "close the replicator -> unload the topic" if the replicator reaches https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java#L190-L192, is it able to fix the issue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171253901
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java:
##########
@@ -268,6 +275,36 @@ protected void readMoreEntries() {
}
}
+ /**
+ * Determine if the messages are continuous, if they are not, the messages are discarded and rewind is called.
+ * Why are the messages discontinuous, for example: if there has something wrong when Replicator is processing
+ * messages "[1:1 ~ 3:3]", Replicator discards the unprocessed message. But a new batch messages "[4:1 ~ 6:6]"
+ * is received later, then these messages will be sent.
+ */
+ protected boolean isMessageContinuousAndRewindIfNot(List<Entry> entries) {
Review Comment:
The reply of this comment is same as https://github.com/apache/pulsar/pull/20128#discussion_r1171219204
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [draft] [fix] [broker] fix messages lost in scenario geo-replication if enabled deduplication [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat closed pull request #20128: [draft] [fix] [broker] fix messages lost in scenario geo-replication if enabled deduplication
URL: https://github.com/apache/pulsar/pull/20128
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#issuecomment-1514075304
> send messages [1,2,3] to the remote cluster, but something is wrong. Replicator trigger rewind and try to send again.
What kind of exceptions will happen to the producer? IMO, we should use infinite publish timeout, if the exception is not publish timeout, it should not be a retriable exception. We should stop the replication.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171218012
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java:
##########
@@ -268,6 +275,36 @@ protected void readMoreEntries() {
}
}
+ /**
+ * Determine if the messages are continuous, if they are not, the messages are discarded and rewind is called.
+ * Why are the messages discontinuous, for example: if there has something wrong when Replicator is processing
+ * messages "[1:1 ~ 3:3]", Replicator discards the unprocessed message. But a new batch messages "[4:1 ~ 6:6]"
+ * is received later, then these messages will be sent.
+ */
+ protected boolean isMessageContinuousAndRewindIfNot(List<Entry> entries) {
Review Comment:
```suggestion
protected boolean hasMessageSkipped(List<Entry> entries) {
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java:
##########
@@ -268,6 +275,36 @@ protected void readMoreEntries() {
}
}
+ /**
+ * Determine if the messages are continuous, if they are not, the messages are discarded and rewind is called.
+ * Why are the messages discontinuous, for example: if there has something wrong when Replicator is processing
+ * messages "[1:1 ~ 3:3]", Replicator discards the unprocessed message. But a new batch messages "[4:1 ~ 6:6]"
+ * is received later, then these messages will be sent.
+ */
+ protected boolean isMessageContinuousAndRewindIfNot(List<Entry> entries) {
+ if (CollectionUtils.isEmpty(entries)){
+ return true;
+ }
+
+ PositionImpl markDeletedPos = (PositionImpl) cursor.getMarkDeletedPosition();
+ PositionImpl lastSentPos = (PositionImpl) lastSent;
+ PositionImpl lastProcessedPos = markDeletedPos.compareTo(lastSentPos) > 0 ? markDeletedPos : lastSentPos;
+
+ PositionImpl firstMessageReceived = (PositionImpl) entries.get(0).getPosition();
+ PositionImpl expectedFirstMessage = managedLedger.getNextValidPosition(lastProcessedPos);
+ if (expectedFirstMessage.compareTo(firstMessageReceived) >= 0) {
+ return true;
+ }
+
+ log.warn("[{}] discontinuous messages was aborted. first message received: {}, expected first message: {}",
Review Comment:
```suggestion
log.warn("[{}] Detected messages are skipped. first received: {}, last processed: {}",
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java:
##########
@@ -268,6 +275,36 @@ protected void readMoreEntries() {
}
}
+ /**
+ * Determine if the messages are continuous, if they are not, the messages are discarded and rewind is called.
+ * Why are the messages discontinuous, for example: if there has something wrong when Replicator is processing
+ * messages "[1:1 ~ 3:3]", Replicator discards the unprocessed message. But a new batch messages "[4:1 ~ 6:6]"
+ * is received later, then these messages will be sent.
+ */
+ protected boolean isMessageContinuousAndRewindIfNot(List<Entry> entries) {
+ if (CollectionUtils.isEmpty(entries)){
+ return true;
+ }
+
+ PositionImpl markDeletedPos = (PositionImpl) cursor.getMarkDeletedPosition();
+ PositionImpl lastSentPos = (PositionImpl) lastSent;
+ PositionImpl lastProcessedPos = markDeletedPos.compareTo(lastSentPos) > 0 ? markDeletedPos : lastSentPos;
+
+ PositionImpl firstMessageReceived = (PositionImpl) entries.get(0).getPosition();
+ PositionImpl expectedFirstMessage = managedLedger.getNextValidPosition(lastProcessedPos);
+ if (expectedFirstMessage.compareTo(firstMessageReceived) >= 0) {
+ return true;
+ }
Review Comment:
It's better to change the method name to `hasMessageSkipped ` because we allow duplicated messages.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java:
##########
@@ -113,6 +117,7 @@ protected boolean replicateEntries(List<Entry> entries) {
// Increment pending messages for messages produced locally
PENDING_MESSAGES_UPDATER.incrementAndGet(this);
+ lastSent = entry.getPosition();
Review Comment:
Same as the comment in GeoReplicator.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java:
##########
@@ -183,6 +188,7 @@ protected boolean replicateEntries(List<Entry> entries) {
msg.getMessageBuilder().clearTxnidLeastBits();
// Increment pending messages for messages produced locally
PENDING_MESSAGES_UPDATER.incrementAndGet(this);
+ lastSent = entry.getPosition();
Review Comment:
The `lastSent` should be updated after the producer sends it successfully.
We are using the async API to send messages.
It looks like the inflight messages will also be treated as `lastSent`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171251110
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ShadowReplicator.java:
##########
@@ -113,6 +117,7 @@ protected boolean replicateEntries(List<Entry> entries) {
// Increment pending messages for messages produced locally
PENDING_MESSAGES_UPDATER.incrementAndGet(this);
+ lastSent = entry.getPosition();
Review Comment:
The reply for this comment is the same as https://github.com/apache/pulsar/pull/20128#discussion_r1171246436
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171246436
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java:
##########
@@ -183,6 +188,7 @@ protected boolean replicateEntries(List<Entry> entries) {
msg.getMessageBuilder().clearTxnidLeastBits();
// Increment pending messages for messages produced locally
PENDING_MESSAGES_UPDATER.incrementAndGet(this);
+ lastSent = entry.getPosition();
Review Comment:
> It looks like the inflight messages will also be treated as lastSent?
Yes, it was done on purpose. Because there has a scenario like this:
- start sending messages `1~3` to the remote cluster, and it is not finished yet(`lastSent` is 0 now).
- receive messages `4~5`. these messages will be discarded and trigger a rewind. This will reduce the efficiency of task Replication.
If we update the variable `lastSent` when the sending is started, it will works like this:
- start sending messages `1~3` to the remote cluster, and it is not finished yet(`lastSent` is 3 now).
- receive messages `4~5` and send them(`lastSent` is 5 now).
- the message `3` failed to send, call `rewind`(`markDeleted` is 2 now) and set the variable `lastSent` to `earliest`.
- Because the timeout of the producer is `0`, the probability of this scenario occurring is very, very small.
- receive the message `3` the second time. because `3 > max(2, earliest)`, the check `hasMessageSkipped` will be pass
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171246436
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java:
##########
@@ -183,6 +188,7 @@ protected boolean replicateEntries(List<Entry> entries) {
msg.getMessageBuilder().clearTxnidLeastBits();
// Increment pending messages for messages produced locally
PENDING_MESSAGES_UPDATER.incrementAndGet(this);
+ lastSent = entry.getPosition();
Review Comment:
> It looks like the inflight messages will also be treated as lastSent?
Yes, it was done on purpose. Because there has a scenario like this:
- start sending messages `1~3` to the remote cluster, and it is not finished yet(`lastSent` is 0 now).
- <strong>(Highlight)</strong>receive messages `4~5`. these messages will be discarded and trigger a rewind. This will reduce the efficiency of task Replication.
If we update the variable `lastSent` when the sending is started, it will works like this:
- start sending messages `1~3` to the remote cluster, and it is not finished yet(`lastSent` is 3 now).
- receive messages `4~5` and send them(`lastSent` is 5 now).
- <strong>(Highlight)</strong>the message `3` failed to send, call `rewind`(`markDeleted` is 2 now) and set the variable `lastSent` to `earliest`.
- Because the timeout of the producer is `0`, the probability of this scenario occurring is very, very small.
- receive the message `3` the second time. because `3 > max(2, earliest)`, the check `hasMessageSkipped` will be pass
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1172025749
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
> This scenario is hard to fix. Because there is no clear marker of which in-flight reads need to be discarded after step-2. I think I should learn how the connector handle messages.
But we will not get an error if the message is duplicated, right? The broker skips the duplicated sends and returns (-1, -1) as the message ID. We are safe to publish messages 1 to 5 again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on a diff in pull request #20128: [fix] [broker] fix messages lost in scenario geo-replication if enabled duplication
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#discussion_r1171415064
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java:
##########
@@ -1756,4 +1763,93 @@ public void testReplicatorProducerNotExceed() throws Exception {
Assert.assertThrows(PulsarClientException.ProducerBusyException.class, () -> new MessageProducer(url2, dest2));
}
+
+ @Test
+ public void testDiscontinuousMessages() throws Exception {
Review Comment:
@codelipenghui
I see:
- for scenario-1: race condition of `read entries` and `rewind`
- a separate PR or PIP will solve this problem
- for scenario-3: https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java#L190-L192
- restart Replicator to fix it.
#### for scenario-2: the task replicate messages is aborted by a loading schema
| time | `receive messages` | `async load schema` |
| --- | --- | --- |
| 1 | receive messages `1~3` |
| 2 | async load schema if needed |
| 3 | receive messages `4~5`|
| 4 | send messages `4~5` to remote cluster |
| 5 | | load schema success, rewind cursor to position `1` |
| 6 | <strong>(Highlight)</strong>Receive messages `1~3` and will send fail by duplication check |
This scenario is hard to fix. Because there is no clear marker of which in-flight reads need to be discarded after step-2. I think I should learn how the connector handle messages.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [draft] [fix] [broker] fix messages lost in scenario geo-replication if enabled deduplication [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on PR #20128:
URL: https://github.com/apache/pulsar/pull/20128#issuecomment-6356566401
Thanks @poorbarcode. I re-checked this against current master. All three scenarios here are now handled at the root, as you and Penghui agreed in review: in-flight reads are fenced on rewind (#24189/#26106), schema fetches and replication errors fence before rewinding, and #23697 switched replication dedup to source positions. Since the approach here was superseded, I'm closing this. If you still see out-of-order replication on master, please open an issue with the sequence and we'll take it from there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org