You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "void-ptr974 (via GitHub)" <gi...@apache.org> on 2026/05/23 04:10:59 UTC
[PR] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
void-ptr974 opened a new pull request, #25860:
URL: https://github.com/apache/pulsar/pull/25860
Fixes #xyz
### Motivation
This fixes a correctness issue in persistent geo-replication V2 deduplication. The issue is in the core replication data path and can produce duplicate messages on the target cluster during normal recovery/failover paths.
Geo-replication V2 deduplication uses the source topic position as the target-side dedup watermark. For a replicated message, the source replicator adds `__MSG_PROP_REPL_SOURCE_POSITION=<source-ledger-id>:<source-entry-id>`, and the target broker stores the latest replicated source position as:
- `<replicator-producer>_LID`
- `<replicator-producer>_EID`
This state is not normal producer sequence state. It is the target-side checkpoint used to identify whether a replayed source entry has already been persisted.
There are three related problems in the current handling of this watermark:
1. **The geo V2 watermark is not recovered from replayed target entries**
When deduplication is enabled or a target topic is loaded, the broker restores dedup state from the `pulsar.dedup` cursor snapshot and then replays entries after that snapshot.
The existing replay path restores normal producer sequence ids from message metadata, but it does not rebuild geo V2 `_LID/_EID` state from replicated messages. If the target topic is unloaded after replicated messages are persisted but before the latest dedup snapshot includes their source positions, the in-memory geo watermark is lost.
If the source replicator later reconnects and replays from its replication cursor, the target broker can fail to identify those source entries as duplicates and persist them again.
2. **The geo V2 watermark is stored as two separate snapshot keys**
A geo V2 watermark is only valid when `_LID` and `_EID` are restored together. Saving only one side does not represent a usable source position.
The snapshot logic previously treated all dedup entries as independent producer states. This makes the geo watermark vulnerable to partial persistence or being crowded out by normal producer sequence state. After reload, an incomplete `_LID/_EID` pair cannot safely deduplicate source replay.
3. **The geo V2 watermark can be removed by normal producer inactivity cleanup**
Normal producer dedup state can be purged after producer inactivity. Geo V2 `_LID/_EID` state has a different lifecycle: it is a source-position watermark for replication, not the lifecycle state of the replicator producer.
The source replicator may disconnect, reconnect, or fail over and replay the same source entries from its replication cursor. If the target has purged the geo watermark because the replicator producer was inactive, those replayed source entries can be accepted and written again.
A concrete failure window is:
1. Source replicator sends messages to the target.
2. Target persists the messages and updates the in-memory geo V2 dedup watermark.
3. The target topic is unloaded before the latest dedup snapshot includes the watermark.
4. The source replicator has not durably advanced its replication cursor yet.
5. Source reconnects and replays the same source entries.
6. Target reloads without the geo watermark and can write duplicates.
This does not require client misuse or corrupted input. Source cursor replay is an expected recovery behavior, so target-side geo dedup state must survive topic reload and producer inactivity cleanup.
### Modifications
- Recover geo-replication V2 dedup watermarks while replaying the dedup cursor by reading `replicatedFrom` and `__MSG_PROP_REPL_SOURCE_POSITION` from persisted replicated messages.
- Store geo-replication watermark keys as complete `_LID/_EID` pairs in dedup snapshots.
- Keep geo-replication watermark state during producer inactivity cleanup, since it represents source position rather than producer lifecycle state.
- Add a regression test that unloads the target topic before the dedup snapshot and forces source replicator replay.
- Add a unit test for geo-replication watermark cleanup behavior.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
./gradlew :pulsar-broker:test --no-configuration-cache \
--tests org.apache.pulsar.broker.service.OneWayReplicatorDeduplicationTest.testGeoReplDedupAfterTargetUnloadBeforeSnapshot \
--tests org.apache.pulsar.broker.service.persistent.MessageDuplicationTest.testInactiveGeoReplicationProducerKeepsDedupState
Result: `BUILD SUCCESSFUL`.
### Does this pull request potentially affect one of the following parts:
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#discussion_r3471601591
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -264,10 +299,30 @@ private CompletableFuture<Void> replayCursor(ManagedCursor cursor) {
highestSequencedPersisted.clear();
inactiveProducers.clear();
// Load the sequence ids from the snapshot in the cursor properties
+ Map<String, Long> replLedgerIds = new HashMap<>();
+ Map<String, Long> replEntryIds = new HashMap<>();
managedCursor.getProperties().forEach((k, v) -> {
- producerRemoved(k);
+ // Geo-replication V2 keys are source-position watermarks, not producer lifecycle state.
+ if (isReplSequenceKey(k)) {
+ String baseProducerName = getBaseProducerName(k);
+ if (k.endsWith(REPL_LEDGER_ID_SUFFIX)) {
+ replLedgerIds.put(baseProducerName, v);
+ } else {
+ replEntryIds.put(baseProducerName, v);
+ }
+ return;
+ }
highestSequencedPushed.put(k, v);
highestSequencedPersisted.put(k, v);
+ producerRemoved(k);
+ });
+ replLedgerIds.forEach((producerName, ledgerId) -> {
+ Long entryId = replEntryIds.get(producerName);
+ if (entryId != null) {
+ ReplSourcePosition position = new ReplSourcePosition(ledgerId, entryId);
+ highestReplPositionPushed.put(producerName, position);
+ highestReplPositionPersisted.put(producerName, position);
+ }
Review Comment:
It is an unexpected situation that `entryId` does not exist. We need to print an error log
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -603,9 +687,17 @@ private CompletableFuture<Void> takeSnapshot(Position position) {
}
Map<String, Long> snapshot = new TreeMap<>();
+ highestReplPositionPersisted.forEach((producerName, replSourcePosition) -> {
+ snapshot.put(producerName + REPL_LEDGER_ID_SUFFIX, replSourcePosition.ledgerId());
+ snapshot.put(producerName + REPL_ENTRY_ID_SUFFIX, replSourcePosition.entryId());
+ });
+ int[] normalProducerCount = new int[1];
Review Comment:
Rather than using `int[1]`, using `int` is better
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -102,6 +103,28 @@ public MessageDupUnknownException(String topicName, String producerName) {
}
}
+ @VisibleForTesting
+ record ReplSourcePosition(long ledgerId, long entryId) implements Comparable<ReplSourcePosition> {
Review Comment:
Using this object enhances the readability and maintainability of the code, but it also increases the GC pressure. I feel that this class is unnecessary
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -102,6 +103,28 @@ public MessageDupUnknownException(String topicName, String producerName) {
}
}
+ @VisibleForTesting
+ record ReplSourcePosition(long ledgerId, long entryId) implements Comparable<ReplSourcePosition> {
+ @Override
+ public int compareTo(ReplSourcePosition other) {
+ int ledgerComparison = Long.compare(ledgerId, other.ledgerId);
+ if (ledgerComparison != 0) {
+ return ledgerComparison;
+ }
+ return Long.compare(entryId, other.entryId);
+ }
+
+ static ReplSourcePosition max(ReplSourcePosition first, ReplSourcePosition second) {
+ return first.compareTo(second) >= 0 ? first : second;
+ }
+ }
+
+ private static class ReplDedupCheck {
Review Comment:
This class seems meaningless; it was just used in the method `isDuplicateReplV2`, using method internal variables can avoid defining a new class
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -356,104 +413,141 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header
MessageMetadata md = Commands.parseMessageMetadata(headersAndPayload);
headersAndPayload.readerIndex(readerIndex);
- List<KeyValue> kvPairList = md.getPropertiesList();
- for (KeyValue kvPair : kvPairList) {
- if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
- if (!kvPair.getValue().contains(":")) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
- }
- String[] ledgerIdAndEntryId = kvPair.getValue().split(":");
- if (ledgerIdAndEntryId.length != 2 || !StringUtils.isNumeric(ledgerIdAndEntryId[0])
- || !StringUtils.isNumeric(ledgerIdAndEntryId[1])) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
+ ReplSourcePosition position = getReplSourcePosition(md, publishContext.getProducerName());
+ if (position != null) {
+ publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION,
+ new long[]{position.ledgerId(), position.entryId()});
+ }
+ }
+ }
+
+ private ReplSourcePosition getReplSourcePosition(MessageMetadata md, String producerName) {
+ List<KeyValue> kvPairList = md.getPropertiesList();
+ for (KeyValue kvPair : kvPairList) {
+ if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
+ String[] ledgerIdAndEntryId = kvPair.getValue().split(":", -1);
+ if (ledgerIdAndEntryId.length != 2) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
+ }
+ try {
+ long ledgerId = Long.parseLong(ledgerIdAndEntryId[0]);
+ long entryId = Long.parseLong(ledgerIdAndEntryId[1]);
+ if (ledgerId >= 0 && entryId >= 0) {
+ return new ReplSourcePosition(ledgerId, entryId);
}
- long[] positionPair = new long[]{Long.valueOf(ledgerIdAndEntryId[0]).longValue(),
- Long.valueOf(ledgerIdAndEntryId[1]).longValue()};
- publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, positionPair);
- break;
+ } catch (NumberFormatException e) {
+ // Log below.
Review Comment:
Seems logging below equals logging here, why not log here?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -356,104 +413,141 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header
MessageMetadata md = Commands.parseMessageMetadata(headersAndPayload);
headersAndPayload.readerIndex(readerIndex);
- List<KeyValue> kvPairList = md.getPropertiesList();
- for (KeyValue kvPair : kvPairList) {
- if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
- if (!kvPair.getValue().contains(":")) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
- }
- String[] ledgerIdAndEntryId = kvPair.getValue().split(":");
- if (ledgerIdAndEntryId.length != 2 || !StringUtils.isNumeric(ledgerIdAndEntryId[0])
- || !StringUtils.isNumeric(ledgerIdAndEntryId[1])) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
+ ReplSourcePosition position = getReplSourcePosition(md, publishContext.getProducerName());
+ if (position != null) {
+ publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION,
+ new long[]{position.ledgerId(), position.entryId()});
+ }
+ }
+ }
+
+ private ReplSourcePosition getReplSourcePosition(MessageMetadata md, String producerName) {
+ List<KeyValue> kvPairList = md.getPropertiesList();
+ for (KeyValue kvPair : kvPairList) {
+ if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
+ String[] ledgerIdAndEntryId = kvPair.getValue().split(":", -1);
+ if (ledgerIdAndEntryId.length != 2) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
+ }
+ try {
+ long ledgerId = Long.parseLong(ledgerIdAndEntryId[0]);
+ long entryId = Long.parseLong(ledgerIdAndEntryId[1]);
+ if (ledgerId >= 0 && entryId >= 0) {
+ return new ReplSourcePosition(ledgerId, entryId);
}
- long[] positionPair = new long[]{Long.valueOf(ledgerIdAndEntryId[0]).longValue(),
- Long.valueOf(ledgerIdAndEntryId[1]).longValue()};
- publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, positionPair);
- break;
+ } catch (NumberFormatException e) {
+ // Log below.
}
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
}
}
+ return null;
}
- public MessageDupStatus isDuplicateReplV2(PublishContext publishContext, ByteBuf headersAndPayload) {
+ private ReplSourcePosition getReplSourcePosition(PublishContext publishContext) {
Object positionPairObj = publishContext.getProperty(MSG_PROP_REPL_SOURCE_POSITION);
- if (positionPairObj == null || !(positionPairObj instanceof long[])) {
- log.error()
- .attr("producerName", publishContext.getProducerName())
- .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
- .attr("sequenceId", publishContext.getSequenceId())
- .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
- .log("Message cannot determine whether it is duplicated due to the acquired message "
- + "props being invalid, prop not in expected format");
- return MessageDupStatus.Unknown;
+ if (!(positionPairObj instanceof long[]) || ((long[]) positionPairObj).length < 2) {
+ return null;
}
-
long[] positionPair = (long[]) positionPairObj;
- long replSequenceLId = positionPair[0];
- long replSequenceEId = positionPair[1];
+ return new ReplSourcePosition(positionPair[0], positionPair[1]);
+ }
- String lastSequenceLIdKey = publishContext.getProducerName() + "_LID";
- String lastSequenceEIdKey = publishContext.getProducerName() + "_EID";
- synchronized (highestSequencedPushed) {
- Long lastSequenceLIdPushed = highestSequencedPushed.get(lastSequenceLIdKey);
- Long lastSequenceEIdPushed = highestSequencedPushed.get(lastSequenceEIdKey);
- if (lastSequenceLIdPushed != null && lastSequenceEIdPushed != null
- && (replSequenceLId < lastSequenceLIdPushed.longValue()
- || (replSequenceLId == lastSequenceLIdPushed.longValue()
- && replSequenceEId <= lastSequenceEIdPushed.longValue()))) {
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPushed", lastSequenceLIdPushed)
- .attr("lastSequenceEIdPushed", lastSequenceEIdPushed)
- .log("Message identified as duplicated");
+ private void logInvalidReplSourcePosition(PublishContext publishContext, String message) {
+ log.error()
+ .attr("producerName", publishContext.getProducerName())
+ .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
+ .attr("sequenceId", publishContext.getSequenceId())
+ .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
+ .log(message);
+ }
- // Also need to check sequence ids that has been persisted.
- // If current message's seq id is smaller or equals to the
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted" than its definitely a dup
- // If current message's seq id is between "lastSequenceLIdPushed:lastSequenceEIdPushed" and
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted", then we cannot be sure whether the message
- // is a dup or not we should return an error to the producer for the latter case so that it can retry
- // at a future time
- Long lastSequenceLIdPersisted = highestSequencedPersisted.get(lastSequenceLIdKey);
- Long lastSequenceEIdPersisted = highestSequencedPersisted.get(lastSequenceEIdKey);
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPersisted", lastSequenceLIdPersisted)
- .attr("lastSequenceEIdPersisted", lastSequenceEIdPersisted)
- .log("Message identified as duplicated producer=. publishing:, latest" + "persisted");
- if (lastSequenceLIdPersisted != null && lastSequenceEIdPersisted != null
- && (replSequenceLId < lastSequenceLIdPersisted.longValue()
- || (replSequenceLId == lastSequenceLIdPersisted.longValue()
- && replSequenceEId <= lastSequenceEIdPersisted))) {
- return MessageDupStatus.Dup;
- } else {
- return MessageDupStatus.Unknown;
- }
+ private void recoverReplWatermark(String producerName, ReplSourcePosition replSourcePosition) {
+ highestReplPositionPushed.merge(producerName, replSourcePosition, ReplSourcePosition::max);
Review Comment:
If sorting is required here, it indicates that there are issues with other codes.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -356,104 +413,141 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header
MessageMetadata md = Commands.parseMessageMetadata(headersAndPayload);
headersAndPayload.readerIndex(readerIndex);
- List<KeyValue> kvPairList = md.getPropertiesList();
- for (KeyValue kvPair : kvPairList) {
- if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
- if (!kvPair.getValue().contains(":")) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
- }
- String[] ledgerIdAndEntryId = kvPair.getValue().split(":");
- if (ledgerIdAndEntryId.length != 2 || !StringUtils.isNumeric(ledgerIdAndEntryId[0])
- || !StringUtils.isNumeric(ledgerIdAndEntryId[1])) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
+ ReplSourcePosition position = getReplSourcePosition(md, publishContext.getProducerName());
+ if (position != null) {
+ publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION,
+ new long[]{position.ledgerId(), position.entryId()});
+ }
+ }
+ }
+
+ private ReplSourcePosition getReplSourcePosition(MessageMetadata md, String producerName) {
+ List<KeyValue> kvPairList = md.getPropertiesList();
+ for (KeyValue kvPair : kvPairList) {
+ if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
+ String[] ledgerIdAndEntryId = kvPair.getValue().split(":", -1);
+ if (ledgerIdAndEntryId.length != 2) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
+ }
+ try {
+ long ledgerId = Long.parseLong(ledgerIdAndEntryId[0]);
+ long entryId = Long.parseLong(ledgerIdAndEntryId[1]);
+ if (ledgerId >= 0 && entryId >= 0) {
+ return new ReplSourcePosition(ledgerId, entryId);
}
- long[] positionPair = new long[]{Long.valueOf(ledgerIdAndEntryId[0]).longValue(),
- Long.valueOf(ledgerIdAndEntryId[1]).longValue()};
- publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, positionPair);
- break;
+ } catch (NumberFormatException e) {
+ // Log below.
}
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
}
}
+ return null;
}
- public MessageDupStatus isDuplicateReplV2(PublishContext publishContext, ByteBuf headersAndPayload) {
+ private ReplSourcePosition getReplSourcePosition(PublishContext publishContext) {
Object positionPairObj = publishContext.getProperty(MSG_PROP_REPL_SOURCE_POSITION);
- if (positionPairObj == null || !(positionPairObj instanceof long[])) {
- log.error()
- .attr("producerName", publishContext.getProducerName())
- .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
- .attr("sequenceId", publishContext.getSequenceId())
- .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
- .log("Message cannot determine whether it is duplicated due to the acquired message "
- + "props being invalid, prop not in expected format");
- return MessageDupStatus.Unknown;
+ if (!(positionPairObj instanceof long[]) || ((long[]) positionPairObj).length < 2) {
+ return null;
}
-
long[] positionPair = (long[]) positionPairObj;
- long replSequenceLId = positionPair[0];
- long replSequenceEId = positionPair[1];
+ return new ReplSourcePosition(positionPair[0], positionPair[1]);
+ }
- String lastSequenceLIdKey = publishContext.getProducerName() + "_LID";
- String lastSequenceEIdKey = publishContext.getProducerName() + "_EID";
- synchronized (highestSequencedPushed) {
- Long lastSequenceLIdPushed = highestSequencedPushed.get(lastSequenceLIdKey);
- Long lastSequenceEIdPushed = highestSequencedPushed.get(lastSequenceEIdKey);
- if (lastSequenceLIdPushed != null && lastSequenceEIdPushed != null
- && (replSequenceLId < lastSequenceLIdPushed.longValue()
- || (replSequenceLId == lastSequenceLIdPushed.longValue()
- && replSequenceEId <= lastSequenceEIdPushed.longValue()))) {
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPushed", lastSequenceLIdPushed)
- .attr("lastSequenceEIdPushed", lastSequenceEIdPushed)
- .log("Message identified as duplicated");
+ private void logInvalidReplSourcePosition(PublishContext publishContext, String message) {
+ log.error()
+ .attr("producerName", publishContext.getProducerName())
+ .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
+ .attr("sequenceId", publishContext.getSequenceId())
+ .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
+ .log(message);
+ }
- // Also need to check sequence ids that has been persisted.
- // If current message's seq id is smaller or equals to the
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted" than its definitely a dup
- // If current message's seq id is between "lastSequenceLIdPushed:lastSequenceEIdPushed" and
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted", then we cannot be sure whether the message
- // is a dup or not we should return an error to the producer for the latter case so that it can retry
- // at a future time
- Long lastSequenceLIdPersisted = highestSequencedPersisted.get(lastSequenceLIdKey);
- Long lastSequenceEIdPersisted = highestSequencedPersisted.get(lastSequenceEIdKey);
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPersisted", lastSequenceLIdPersisted)
- .attr("lastSequenceEIdPersisted", lastSequenceEIdPersisted)
- .log("Message identified as duplicated producer=. publishing:, latest" + "persisted");
- if (lastSequenceLIdPersisted != null && lastSequenceEIdPersisted != null
- && (replSequenceLId < lastSequenceLIdPersisted.longValue()
- || (replSequenceLId == lastSequenceLIdPersisted.longValue()
- && replSequenceEId <= lastSequenceEIdPersisted))) {
- return MessageDupStatus.Dup;
- } else {
- return MessageDupStatus.Unknown;
- }
+ private void recoverReplWatermark(String producerName, ReplSourcePosition replSourcePosition) {
+ highestReplPositionPushed.merge(producerName, replSourcePosition, ReplSourcePosition::max);
+ highestReplPositionPersisted.merge(producerName, replSourcePosition, ReplSourcePosition::max);
+ }
+
+ private void recordReplWatermarkPersisted(String producerName, ReplSourcePosition replSourcePosition) {
+ highestReplPositionPersisted.merge(producerName, replSourcePosition, ReplSourcePosition::max);
+ }
+
+ private boolean isReplPositionAtOrBefore(ReplSourcePosition position, ReplSourcePosition watermark) {
+ return position.compareTo(watermark) <= 0;
+ }
+
+ private void logUnexpectedReplSourcePosition(String producerName, String value) {
+ log.warn()
+ .attr("producerName", producerName)
+ .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
+ .attr("value", value)
+ .log("Unexpected");
+ }
+
+ @VisibleForTesting
+ void recoverReplWatermarkFromMetadata(MessageMetadata md) {
+ if (md.hasMarkerType()) {
+ return;
+ }
+ String replProducerName = getReplProducerName(md);
+ if (replProducerName != null) {
+ ReplSourcePosition replSourcePosition = getReplSourcePosition(md, replProducerName);
+ if (replSourcePosition != null) {
+ recoverReplWatermark(replProducerName, replSourcePosition);
}
- highestSequencedPushed.put(lastSequenceLIdKey, replSequenceLId);
- highestSequencedPushed.put(lastSequenceEIdKey, replSequenceEId);
}
+ }
+
+ @VisibleForTesting
+ String getReplProducerName(MessageMetadata md) {
+ final var shadowSourceTopic = topic.getShadowSourceTopic();
+ if (md.hasReplicatedFrom() && shadowSourceTopic.isPresent()) {
+ return ShadowReplicator.getShadowProducerName(replicatorPrefix,
+ shadowSourceTopic.get().toString(), topic.getName());
+ }
+ if (md.hasReplicatedFrom()) {
+ return AbstractReplicator.getReplicatorName(replicatorPrefix, md.getReplicatedFrom())
+ + AbstractReplicator.REPL_PRODUCER_NAME_DELIMITER
+ + pulsar.getConfiguration().getClusterName();
+ }
+ if (Producer.isRemoteOrShadow(md.getProducerName(), replicatorPrefix)) {
+ return md.getProducerName();
+ }
+ return null;
+ }
+
+ public MessageDupStatus isDuplicateReplV2(PublishContext publishContext, ByteBuf headersAndPayload) {
+ ReplSourcePosition replSourcePosition = getReplSourcePosition(publishContext);
+ if (replSourcePosition == null) {
+ logInvalidReplSourcePosition(publishContext, "Message cannot determine whether it is duplicated due to "
+ + "the acquired message props being invalid, prop not in expected format");
+ return MessageDupStatus.Unknown;
+ }
+
+ String producerName = publishContext.getProducerName();
+ ReplDedupCheck check = new ReplDedupCheck();
+ highestReplPositionPushed.compute(producerName, (__, lastPushed) -> {
Review Comment:
Almost all the changes were made around the producer's name:
**Before**
- Cached `{producerName} + suffix`
- Stored `{producerName} + suffix`
**After**
- Cached `{producerName}`
- Stored `{producerName} + suffix`
I consider the new solution to be better, which improves the performance.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -276,6 +331,8 @@ private CompletableFuture<Void> replayCursor(ManagedCursor cursor) {
return replayTask.replay(cursor, (__, buffer) -> {
final var metadata = Commands.parseMessageMetadata(buffer);
final var producerName = metadata.getProducerName();
+ // Rebuild replication watermarks from entries written after the last dedup snapshot.
+ recoverReplWatermarkFromMetadata(metadata);
final var sequenceId = Math.max(metadata.getHighestSequenceId(), metadata.getSequenceId());
Review Comment:
Since we already use the new collections `highestReplPositionPushed` and `highestReplPositionPersisted` to trace the deduplication key, we should not put them into `highestSequencedPushed` and `highestSequencedPersisted` anymore.
- `is marker`: it should be traced by `highestSequencedPushed` and `highestSequencedPersisted`
- otherwise: it should be traced by `highestReplPositionPushed` and `highestReplPositionPersisted`
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java:
##########
@@ -272,6 +273,107 @@ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
admin2.topics().unload(topicName);
}
+ @Test
+ public void testGeoReplDedupAfterTargetUnloadBeforeSnapshot() throws Exception {
+ // Drop send receipts so the source replication cursor cannot advance while target messages are persisted.
+ AtomicBoolean stuckSendReceipt = new AtomicBoolean(true);
+ Runnable cleanInjection = injectReplicatorClientCnx((conf, eventLoopGroup) -> new ClientCnx(
+ InstrumentProvider.NOOP, conf, eventLoopGroup) {
+ @Override
+ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
+ if (!stuckSendReceipt.get()) {
+ super.handleSendReceipt(sendReceipt);
+ }
+ }
+ });
+
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + nonReplicatedNamespace + "/tp_");
+ final String subscription = "s1";
+ Producer<Integer> producer = null;
+ Consumer<Integer> consumer = null;
+ try {
+ admin1.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createSubscription(topicName, subscription, MessageId.earliest);
+
+ PersistentTopic persistentTopic2 =
+ (PersistentTopic) pulsar2.getBrokerService().getTopic(topicName, false).join().get();
+ admin2.topicPolicies().setDeduplicationStatus(topicName, true);
+ Awaitility.await().untilAsserted(() -> {
+ MessageDeduplication messageDeduplication2 = persistentTopic2.getMessageDeduplication();
+ assertEquals(String.valueOf(messageDeduplication2.getStatus()), "Enabled");
+ });
+
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1, cluster2));
+ waitReplicatorStarted(topicName);
+ PersistentTopic persistentTopic1 =
+ (PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
+
+ producer = client1.newProducer(Schema.INT32).topic(topicName).create();
+ int messageCount = 4;
+ for (int i = 0; i < messageCount; i++) {
+ producer.send(i);
+ }
+
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertEquals(admin2.topics().getStats(topicName).getMsgInCounter(), messageCount);
+ });
+
+ GeoPersistentReplicator replicator =
+ (GeoPersistentReplicator) persistentTopic1.getReplicators().get(cluster2);
+ assertNotNull(replicator);
+ AtomicReference<ClientCnx> replicatorClientCnx = new AtomicReference<>();
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertNotNull(replicator.producer);
+ ClientCnx clientCnx = replicator.producer.getClientCnx();
+ assertNotNull(clientCnx);
+ replicatorClientCnx.set(clientCnx);
+ });
+
+ // Reload target before a dedup snapshot has to cover the latest replicated source positions.
+ admin2.topics().unload(topicName);
+ pulsar2.getBrokerService().getTopic(topicName, false).join();
+
+ // Reconnect the source replicator. It replays from the old replication cursor and target must deduplicate.
+ stuckSendReceipt.set(false);
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertNotNull(replicator.producer);
+ ClientCnx clientCnx = replicator.producer.getClientCnx();
+ assertNotNull(clientCnx);
+ replicatorClientCnx.set(clientCnx);
+ });
+ replicatorClientCnx.get().ctx().channel().close();
+
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertEquals(replicator.getCursor().getNumberOfEntriesInBacklog(true), 0);
Review Comment:
Since you called `topics unload`, the object replicator will be rebuilt; the verification here is meaningless
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java:
##########
@@ -272,6 +273,107 @@ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
admin2.topics().unload(topicName);
}
+ @Test
+ public void testGeoReplDedupAfterTargetUnloadBeforeSnapshot() throws Exception {
+ // Drop send receipts so the source replication cursor cannot advance while target messages are persisted.
+ AtomicBoolean stuckSendReceipt = new AtomicBoolean(true);
+ Runnable cleanInjection = injectReplicatorClientCnx((conf, eventLoopGroup) -> new ClientCnx(
+ InstrumentProvider.NOOP, conf, eventLoopGroup) {
+ @Override
+ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
+ if (!stuckSendReceipt.get()) {
+ super.handleSendReceipt(sendReceipt);
+ }
+ }
+ });
+
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + nonReplicatedNamespace + "/tp_");
+ final String subscription = "s1";
+ Producer<Integer> producer = null;
+ Consumer<Integer> consumer = null;
+ try {
+ admin1.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createSubscription(topicName, subscription, MessageId.earliest);
+
+ PersistentTopic persistentTopic2 =
+ (PersistentTopic) pulsar2.getBrokerService().getTopic(topicName, false).join().get();
+ admin2.topicPolicies().setDeduplicationStatus(topicName, true);
+ Awaitility.await().untilAsserted(() -> {
+ MessageDeduplication messageDeduplication2 = persistentTopic2.getMessageDeduplication();
+ assertEquals(String.valueOf(messageDeduplication2.getStatus()), "Enabled");
+ });
+
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1, cluster2));
+ waitReplicatorStarted(topicName);
+ PersistentTopic persistentTopic1 =
+ (PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
+
+ producer = client1.newProducer(Schema.INT32).topic(topicName).create();
+ int messageCount = 4;
+ for (int i = 0; i < messageCount; i++) {
+ producer.send(i);
+ }
+
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertEquals(admin2.topics().getStats(topicName).getMsgInCounter(), messageCount);
+ });
+
+ GeoPersistentReplicator replicator =
+ (GeoPersistentReplicator) persistentTopic1.getReplicators().get(cluster2);
+ assertNotNull(replicator);
+ AtomicReference<ClientCnx> replicatorClientCnx = new AtomicReference<>();
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertNotNull(replicator.producer);
+ ClientCnx clientCnx = replicator.producer.getClientCnx();
+ assertNotNull(clientCnx);
+ replicatorClientCnx.set(clientCnx);
+ });
+
+ // Reload target before a dedup snapshot has to cover the latest replicated source positions.
+ admin2.topics().unload(topicName);
+ pulsar2.getBrokerService().getTopic(topicName, false).join();
+
+ // Reconnect the source replicator. It replays from the old replication cursor and target must deduplicate.
+ stuckSendReceipt.set(false);
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertNotNull(replicator.producer);
Review Comment:
Since you called `topics unload`, the object `replicator` will be rebuilt; the verification here is meaningless
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -356,104 +413,141 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header
MessageMetadata md = Commands.parseMessageMetadata(headersAndPayload);
headersAndPayload.readerIndex(readerIndex);
- List<KeyValue> kvPairList = md.getPropertiesList();
- for (KeyValue kvPair : kvPairList) {
- if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
- if (!kvPair.getValue().contains(":")) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
- }
- String[] ledgerIdAndEntryId = kvPair.getValue().split(":");
- if (ledgerIdAndEntryId.length != 2 || !StringUtils.isNumeric(ledgerIdAndEntryId[0])
- || !StringUtils.isNumeric(ledgerIdAndEntryId[1])) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
+ ReplSourcePosition position = getReplSourcePosition(md, publishContext.getProducerName());
+ if (position != null) {
+ publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION,
+ new long[]{position.ledgerId(), position.entryId()});
+ }
+ }
+ }
+
+ private ReplSourcePosition getReplSourcePosition(MessageMetadata md, String producerName) {
+ List<KeyValue> kvPairList = md.getPropertiesList();
+ for (KeyValue kvPair : kvPairList) {
+ if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
+ String[] ledgerIdAndEntryId = kvPair.getValue().split(":", -1);
+ if (ledgerIdAndEntryId.length != 2) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
+ }
+ try {
+ long ledgerId = Long.parseLong(ledgerIdAndEntryId[0]);
+ long entryId = Long.parseLong(ledgerIdAndEntryId[1]);
+ if (ledgerId >= 0 && entryId >= 0) {
+ return new ReplSourcePosition(ledgerId, entryId);
}
- long[] positionPair = new long[]{Long.valueOf(ledgerIdAndEntryId[0]).longValue(),
- Long.valueOf(ledgerIdAndEntryId[1]).longValue()};
- publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, positionPair);
- break;
+ } catch (NumberFormatException e) {
+ // Log below.
}
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
}
}
+ return null;
}
- public MessageDupStatus isDuplicateReplV2(PublishContext publishContext, ByteBuf headersAndPayload) {
+ private ReplSourcePosition getReplSourcePosition(PublishContext publishContext) {
Object positionPairObj = publishContext.getProperty(MSG_PROP_REPL_SOURCE_POSITION);
- if (positionPairObj == null || !(positionPairObj instanceof long[])) {
- log.error()
- .attr("producerName", publishContext.getProducerName())
- .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
- .attr("sequenceId", publishContext.getSequenceId())
- .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
- .log("Message cannot determine whether it is duplicated due to the acquired message "
- + "props being invalid, prop not in expected format");
- return MessageDupStatus.Unknown;
+ if (!(positionPairObj instanceof long[]) || ((long[]) positionPairObj).length < 2) {
+ return null;
}
-
long[] positionPair = (long[]) positionPairObj;
- long replSequenceLId = positionPair[0];
- long replSequenceEId = positionPair[1];
+ return new ReplSourcePosition(positionPair[0], positionPair[1]);
+ }
- String lastSequenceLIdKey = publishContext.getProducerName() + "_LID";
- String lastSequenceEIdKey = publishContext.getProducerName() + "_EID";
- synchronized (highestSequencedPushed) {
- Long lastSequenceLIdPushed = highestSequencedPushed.get(lastSequenceLIdKey);
- Long lastSequenceEIdPushed = highestSequencedPushed.get(lastSequenceEIdKey);
- if (lastSequenceLIdPushed != null && lastSequenceEIdPushed != null
- && (replSequenceLId < lastSequenceLIdPushed.longValue()
- || (replSequenceLId == lastSequenceLIdPushed.longValue()
- && replSequenceEId <= lastSequenceEIdPushed.longValue()))) {
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPushed", lastSequenceLIdPushed)
- .attr("lastSequenceEIdPushed", lastSequenceEIdPushed)
- .log("Message identified as duplicated");
+ private void logInvalidReplSourcePosition(PublishContext publishContext, String message) {
+ log.error()
+ .attr("producerName", publishContext.getProducerName())
+ .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
+ .attr("sequenceId", publishContext.getSequenceId())
+ .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
+ .log(message);
+ }
- // Also need to check sequence ids that has been persisted.
- // If current message's seq id is smaller or equals to the
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted" than its definitely a dup
- // If current message's seq id is between "lastSequenceLIdPushed:lastSequenceEIdPushed" and
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted", then we cannot be sure whether the message
- // is a dup or not we should return an error to the producer for the latter case so that it can retry
- // at a future time
- Long lastSequenceLIdPersisted = highestSequencedPersisted.get(lastSequenceLIdKey);
- Long lastSequenceEIdPersisted = highestSequencedPersisted.get(lastSequenceEIdKey);
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPersisted", lastSequenceLIdPersisted)
- .attr("lastSequenceEIdPersisted", lastSequenceEIdPersisted)
- .log("Message identified as duplicated producer=. publishing:, latest" + "persisted");
- if (lastSequenceLIdPersisted != null && lastSequenceEIdPersisted != null
- && (replSequenceLId < lastSequenceLIdPersisted.longValue()
- || (replSequenceLId == lastSequenceLIdPersisted.longValue()
- && replSequenceEId <= lastSequenceEIdPersisted))) {
- return MessageDupStatus.Dup;
- } else {
- return MessageDupStatus.Unknown;
- }
+ private void recoverReplWatermark(String producerName, ReplSourcePosition replSourcePosition) {
+ highestReplPositionPushed.merge(producerName, replSourcePosition, ReplSourcePosition::max);
+ highestReplPositionPersisted.merge(producerName, replSourcePosition, ReplSourcePosition::max);
+ }
+
+ private void recordReplWatermarkPersisted(String producerName, ReplSourcePosition replSourcePosition) {
+ highestReplPositionPersisted.merge(producerName, replSourcePosition, ReplSourcePosition::max);
Review Comment:
If sorting is required here, it indicates that there are issues with other codes.
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#issuecomment-4801785020
Pushed `e823f0a01e8` with the review updates:
- Split geo/shadow replication watermark recovery from normal producer sequence recovery during replay.
- Added error logging for incomplete geo-replication watermark snapshot pairs.
- Removed the extra `ReplDedupCheck` helper and simplified `isDuplicateReplV2` local state handling.
- Removed the watermark ordering/max merge logic.
- Changed snapshot normal-producer counting to use a plain `int` loop.
- Updated inactive geo-replication cleanup logging to match the actual behavior.
- Updated the unload-before-snapshot test to re-fetch the current source topic/replicator and wait for target deduplication to become `Enabled`.
- Avoided adding new reflection in the new unit tests by using testing accessors.
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#discussion_r4273437359
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -264,10 +299,30 @@ private CompletableFuture<Void> replayCursor(ManagedCursor cursor) {
highestSequencedPersisted.clear();
inactiveProducers.clear();
// Load the sequence ids from the snapshot in the cursor properties
+ Map<String, Long> replLedgerIds = new HashMap<>();
+ Map<String, Long> replEntryIds = new HashMap<>();
managedCursor.getProperties().forEach((k, v) -> {
- producerRemoved(k);
+ // Geo-replication V2 keys are source-position watermarks, not producer lifecycle state.
+ if (isReplSequenceKey(k)) {
+ String baseProducerName = getBaseProducerName(k);
+ if (k.endsWith(REPL_LEDGER_ID_SUFFIX)) {
+ replLedgerIds.put(baseProducerName, v);
+ } else {
+ replEntryIds.put(baseProducerName, v);
+ }
+ return;
+ }
highestSequencedPushed.put(k, v);
highestSequencedPersisted.put(k, v);
+ producerRemoved(k);
+ });
+ replLedgerIds.forEach((producerName, ledgerId) -> {
+ Long entryId = replEntryIds.get(producerName);
+ if (entryId != null) {
+ ReplSourcePosition position = new ReplSourcePosition(ledgerId, entryId);
+ highestReplPositionPushed.put(producerName, position);
+ highestReplPositionPersisted.put(producerName, position);
+ }
Review Comment:
I need to correct my earlier reply about logging this as an error. Legacy snapshots use a flat key space, so a valid V1 producer name ending in `_LID` or `_EID` can appear to have a missing counterpart. Recovery preserves its sequence state, restores a replication watermark only for a complete pair, and logs the ambiguous case at DEBUG. `testSnapshotRetainsProducerNameEndingWithReplicationPositionSuffix` covers both suffixes.
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#discussion_r3989284142
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -262,24 +278,54 @@ private CompletableFuture<Void> replayCursor(ManagedCursor cursor) {
snapshotCounter = 0;
highestSequencedPushed.clear();
highestSequencedPersisted.clear();
+ highestReplPositionPushed.clear();
+ highestReplPositionPersisted.clear();
inactiveProducers.clear();
// Load the sequence ids from the snapshot in the cursor properties
+ Map<String, Long> replLedgerIds = new HashMap<>();
+ Map<String, Long> replEntryIds = new HashMap<>();
managedCursor.getProperties().forEach((k, v) -> {
- producerRemoved(k);
+ // Geo-replication V2 keys are source-position watermarks, not producer lifecycle state.
+ if (isReplSequenceKey(k)) {
+ String baseProducerName = getBaseProducerName(k);
+ if (k.endsWith(REPL_LEDGER_ID_SUFFIX)) {
+ replLedgerIds.put(baseProducerName, v);
+ } else {
+ replEntryIds.put(baseProducerName, v);
+ }
+ return;
+ }
highestSequencedPushed.put(k, v);
highestSequencedPersisted.put(k, v);
+ producerRemoved(k);
+ });
+ replLedgerIds.forEach((producerName, ledgerId) -> {
+ Long entryId = replEntryIds.get(producerName);
+ if (entryId != null) {
+ ReplSourcePosition position = new ReplSourcePosition(ledgerId, entryId);
+ highestReplPositionPushed.put(producerName, position);
+ highestReplPositionPersisted.put(producerName, position);
+ } else {
+ logIncompleteReplSourcePositionSnapshot(producerName, ledgerId, null);
+ }
+ });
+ replEntryIds.forEach((producerName, entryId) -> {
+ if (!replLedgerIds.containsKey(producerName)) {
+ logIncompleteReplSourcePositionSnapshot(producerName, null, entryId);
+ }
});
// Replay all the entries and apply all the sequence ids updates
log.info()
.attr("numberOfEntries", managedCursor.getNumberOfEntries())
.log("Replaying entries for deduplication");
return replayTask.replay(cursor, (__, buffer) -> {
final var metadata = Commands.parseMessageMetadata(buffer);
- final var producerName = metadata.getProducerName();
- final var sequenceId = Math.max(metadata.getHighestSequenceId(), metadata.getSequenceId());
- highestSequencedPushed.put(producerName, sequenceId);
- highestSequencedPersisted.put(producerName, sequenceId);
- producerRemoved(producerName);
+ if (metadata.hasMarkerType() || getReplProducerName(metadata) == null) {
+ recoverProducerSequenceIdFromMetadata(metadata);
+ } else {
+ // Rebuild replication watermarks from entries written after the last dedup snapshot.
+ recoverReplWatermarkFromMetadata(metadata);
Review Comment:
Fixed in 7cab7835ed4. Replay now restores a V2 watermark only when a non-marker replicated entry has a valid source-position property; otherwise it falls back to the original producer sequence recovery required by V1. I also added a persisted-entry unload/reload regression in a2c8baee6e1: it writes a V1 replicated entry after the snapshot, unloads and reloads the topic, verifies recovery under the original producer name, and verifies that replay is classified as a duplicate. The mixed V1/V2/invalid replay cases and the additional V2 state-transition coverage are included as well. The full MessageDuplicationTest class and broker checkstyle passed locally. Could you please take another look?
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#discussion_r4273434127
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -356,104 +402,179 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header
MessageMetadata md = Commands.parseMessageMetadata(headersAndPayload);
headersAndPayload.readerIndex(readerIndex);
- List<KeyValue> kvPairList = md.getPropertiesList();
- for (KeyValue kvPair : kvPairList) {
- if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
- if (!kvPair.getValue().contains(":")) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
- }
- String[] ledgerIdAndEntryId = kvPair.getValue().split(":");
- if (ledgerIdAndEntryId.length != 2 || !StringUtils.isNumeric(ledgerIdAndEntryId[0])
- || !StringUtils.isNumeric(ledgerIdAndEntryId[1])) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
+ long[] position = getReplSourcePosition(md, publishContext.getProducerName());
+ if (position != null) {
+ publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, position);
+ }
+ }
+ }
+
+ private long[] getReplSourcePosition(MessageMetadata md, String producerName) {
+ List<KeyValue> kvPairList = md.getPropertiesList();
+ for (KeyValue kvPair : kvPairList) {
+ if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
+ String[] ledgerIdAndEntryId = kvPair.getValue().split(":", -1);
+ if (ledgerIdAndEntryId.length != 2) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
+ }
+ try {
+ long ledgerId = Long.parseLong(ledgerIdAndEntryId[0]);
+ long entryId = Long.parseLong(ledgerIdAndEntryId[1]);
+ if (ledgerId >= 0 && entryId >= 0) {
+ return new long[]{ledgerId, entryId};
}
- long[] positionPair = new long[]{Long.valueOf(ledgerIdAndEntryId[0]).longValue(),
- Long.valueOf(ledgerIdAndEntryId[1]).longValue()};
- publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, positionPair);
- break;
+ } catch (NumberFormatException e) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
}
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
}
}
+ return null;
}
- public MessageDupStatus isDuplicateReplV2(PublishContext publishContext, ByteBuf headersAndPayload) {
+ private long[] getReplSourcePosition(PublishContext publishContext) {
Object positionPairObj = publishContext.getProperty(MSG_PROP_REPL_SOURCE_POSITION);
- if (positionPairObj == null || !(positionPairObj instanceof long[])) {
- log.error()
- .attr("producerName", publishContext.getProducerName())
- .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
- .attr("sequenceId", publishContext.getSequenceId())
- .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
- .log("Message cannot determine whether it is duplicated due to the acquired message "
- + "props being invalid, prop not in expected format");
- return MessageDupStatus.Unknown;
+ if (!(positionPairObj instanceof long[]) || ((long[]) positionPairObj).length < 2) {
+ return null;
}
+ return (long[]) positionPairObj;
+ }
- long[] positionPair = (long[]) positionPairObj;
- long replSequenceLId = positionPair[0];
- long replSequenceEId = positionPair[1];
+ private void logInvalidReplSourcePosition(PublishContext publishContext, String message) {
+ log.error()
+ .attr("producerName", publishContext.getProducerName())
+ .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
+ .attr("sequenceId", publishContext.getSequenceId())
+ .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
+ .log(message);
+ }
- String lastSequenceLIdKey = publishContext.getProducerName() + "_LID";
- String lastSequenceEIdKey = publishContext.getProducerName() + "_EID";
- synchronized (highestSequencedPushed) {
- Long lastSequenceLIdPushed = highestSequencedPushed.get(lastSequenceLIdKey);
- Long lastSequenceEIdPushed = highestSequencedPushed.get(lastSequenceEIdKey);
- if (lastSequenceLIdPushed != null && lastSequenceEIdPushed != null
- && (replSequenceLId < lastSequenceLIdPushed.longValue()
- || (replSequenceLId == lastSequenceLIdPushed.longValue()
- && replSequenceEId <= lastSequenceEIdPushed.longValue()))) {
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPushed", lastSequenceLIdPushed)
- .attr("lastSequenceEIdPushed", lastSequenceEIdPushed)
- .log("Message identified as duplicated");
+ private void recoverReplWatermark(String producerName, long ledgerId, long entryId) {
+ ReplSourcePosition replSourcePosition = new ReplSourcePosition(ledgerId, entryId);
+ highestReplPositionPushed.merge(producerName, replSourcePosition, this::maxReplPosition);
+ highestReplPositionPersisted.merge(producerName, replSourcePosition, this::maxReplPosition);
+ }
- // Also need to check sequence ids that has been persisted.
- // If current message's seq id is smaller or equals to the
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted" than its definitely a dup
- // If current message's seq id is between "lastSequenceLIdPushed:lastSequenceEIdPushed" and
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted", then we cannot be sure whether the message
- // is a dup or not we should return an error to the producer for the latter case so that it can retry
- // at a future time
- Long lastSequenceLIdPersisted = highestSequencedPersisted.get(lastSequenceLIdKey);
- Long lastSequenceEIdPersisted = highestSequencedPersisted.get(lastSequenceEIdKey);
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPersisted", lastSequenceLIdPersisted)
- .attr("lastSequenceEIdPersisted", lastSequenceEIdPersisted)
- .log("Message identified as duplicated producer=. publishing:, latest" + "persisted");
- if (lastSequenceLIdPersisted != null && lastSequenceEIdPersisted != null
- && (replSequenceLId < lastSequenceLIdPersisted.longValue()
- || (replSequenceLId == lastSequenceLIdPersisted.longValue()
- && replSequenceEId <= lastSequenceEIdPersisted))) {
- return MessageDupStatus.Dup;
- } else {
- return MessageDupStatus.Unknown;
- }
+ private void recordReplWatermarkPersisted(String producerName, long ledgerId, long entryId) {
+ highestReplPositionPersisted.merge(producerName, new ReplSourcePosition(ledgerId, entryId),
+ this::maxReplPosition);
+ }
+
+ private boolean isReplPositionAtOrBefore(long ledgerId, long entryId, ReplSourcePosition watermark) {
+ return ledgerId < watermark.ledgerId()
+ || (ledgerId == watermark.ledgerId() && entryId <= watermark.entryId());
+ }
+
+ private ReplSourcePosition maxReplPosition(ReplSourcePosition current, ReplSourcePosition candidate) {
+ return isReplPositionAtOrBefore(candidate.ledgerId(), candidate.entryId(), current) ? current : candidate;
+ }
+
+ private void logUnexpectedReplSourcePosition(String producerName, String value) {
+ log.warn()
+ .attr("producerName", producerName)
+ .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
+ .attr("value", value)
+ .log("Unexpected");
+ }
+
+ private void logIncompleteReplSourcePositionSnapshot(String producerName, Long ledgerId, Long entryId) {
+ log.error()
+ .attr("producerName", producerName)
+ .attr("ledgerId", ledgerId)
+ .attr("entryId", entryId)
+ .log("Incomplete geo-replication deduplication snapshot");
+ }
+
+ private void recoverProducerSequenceIdFromMetadata(MessageMetadata md) {
+ final var producerName = md.getProducerName();
+ final var sequenceId = Math.max(md.getHighestSequenceId(), md.getSequenceId());
+ highestSequencedPushed.merge(producerName, sequenceId, Math::max);
Review Comment:
Thanks, you're right. Replay now uses `put` for both producer sequences and replication watermarks, so the last entry in ledger order wins. `merge(max)` remains only for live persistence callbacks, which can complete out of order. I added `testReplayRecoversSequenceIdAfterInactiveProducerNameIsReused`: after sequence 100 is purged and the same name writes sequence 0, unload/reload accepts sequence 1 and still deduplicates sequence 0. Could you take another look?
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#discussion_r3977567275
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -262,24 +278,54 @@ private CompletableFuture<Void> replayCursor(ManagedCursor cursor) {
snapshotCounter = 0;
highestSequencedPushed.clear();
highestSequencedPersisted.clear();
+ highestReplPositionPushed.clear();
+ highestReplPositionPersisted.clear();
inactiveProducers.clear();
// Load the sequence ids from the snapshot in the cursor properties
+ Map<String, Long> replLedgerIds = new HashMap<>();
+ Map<String, Long> replEntryIds = new HashMap<>();
managedCursor.getProperties().forEach((k, v) -> {
- producerRemoved(k);
+ // Geo-replication V2 keys are source-position watermarks, not producer lifecycle state.
+ if (isReplSequenceKey(k)) {
+ String baseProducerName = getBaseProducerName(k);
+ if (k.endsWith(REPL_LEDGER_ID_SUFFIX)) {
+ replLedgerIds.put(baseProducerName, v);
+ } else {
+ replEntryIds.put(baseProducerName, v);
+ }
+ return;
+ }
highestSequencedPushed.put(k, v);
highestSequencedPersisted.put(k, v);
+ producerRemoved(k);
+ });
+ replLedgerIds.forEach((producerName, ledgerId) -> {
+ Long entryId = replEntryIds.get(producerName);
+ if (entryId != null) {
+ ReplSourcePosition position = new ReplSourcePosition(ledgerId, entryId);
+ highestReplPositionPushed.put(producerName, position);
+ highestReplPositionPersisted.put(producerName, position);
+ } else {
+ logIncompleteReplSourcePositionSnapshot(producerName, ledgerId, null);
+ }
+ });
+ replEntryIds.forEach((producerName, entryId) -> {
+ if (!replLedgerIds.containsKey(producerName)) {
+ logIncompleteReplSourcePositionSnapshot(producerName, null, entryId);
+ }
});
// Replay all the entries and apply all the sequence ids updates
log.info()
.attr("numberOfEntries", managedCursor.getNumberOfEntries())
.log("Replaying entries for deduplication");
return replayTask.replay(cursor, (__, buffer) -> {
final var metadata = Commands.parseMessageMetadata(buffer);
- final var producerName = metadata.getProducerName();
- final var sequenceId = Math.max(metadata.getHighestSequenceId(), metadata.getSequenceId());
- highestSequencedPushed.put(producerName, sequenceId);
- highestSequencedPersisted.put(producerName, sequenceId);
- producerRemoved(producerName);
+ if (metadata.hasMarkerType() || getReplProducerName(metadata) == null) {
+ recoverProducerSequenceIdFromMetadata(metadata);
+ } else {
+ // Rebuild replication watermarks from entries written after the last dedup snapshot.
+ recoverReplWatermarkFromMetadata(metadata);
Review Comment:
**[BUG] Preserve V1 producer sequence state during replay**
Messages from an older source broker have `replicatedFrom` but do not have `__MSG_PROP_REPL_SOURCE_POSITION`. They now enter this branch because `getReplProducerName` returns a name, and `recoverReplWatermarkFromMetadata` then does nothing when the property is absent. This skips the original producer sequence recovery that V1 still needs: `isDuplicateReplV1` uses the original producer name and `recordMessagePersistedNormal` records that producer’s sequence state ([`MessageDeduplication.java:369-382`](https://github.com/apache/pulsar/blob/5ab9479514e4a0cbf1fd379e1e184acf6e903280/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java#L369-L382) and [`MessageDeduplication.java:648-660`](https://github.com/apache/pulsar/blob/5ab9479514e4a0cbf1fd379e1e184acf6e903280/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java#L648-L660)).
If the target persists V1 messages after the last dedup snapshot, unloads, and the older source reconnects and replays them, the recovered normal-producer watermark is missing or stale and those messages are persisted again. Keep the normal sequence recovery for V1 entries, including entries without a V2 position, and add an unload-before-snapshot regression with a V1 source.
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#discussion_r4254853681
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -356,104 +402,179 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header
MessageMetadata md = Commands.parseMessageMetadata(headersAndPayload);
headersAndPayload.readerIndex(readerIndex);
- List<KeyValue> kvPairList = md.getPropertiesList();
- for (KeyValue kvPair : kvPairList) {
- if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
- if (!kvPair.getValue().contains(":")) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
- }
- String[] ledgerIdAndEntryId = kvPair.getValue().split(":");
- if (ledgerIdAndEntryId.length != 2 || !StringUtils.isNumeric(ledgerIdAndEntryId[0])
- || !StringUtils.isNumeric(ledgerIdAndEntryId[1])) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
+ long[] position = getReplSourcePosition(md, publishContext.getProducerName());
+ if (position != null) {
+ publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, position);
+ }
+ }
+ }
+
+ private long[] getReplSourcePosition(MessageMetadata md, String producerName) {
+ List<KeyValue> kvPairList = md.getPropertiesList();
+ for (KeyValue kvPair : kvPairList) {
+ if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
+ String[] ledgerIdAndEntryId = kvPair.getValue().split(":", -1);
+ if (ledgerIdAndEntryId.length != 2) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
+ }
+ try {
+ long ledgerId = Long.parseLong(ledgerIdAndEntryId[0]);
+ long entryId = Long.parseLong(ledgerIdAndEntryId[1]);
+ if (ledgerId >= 0 && entryId >= 0) {
+ return new long[]{ledgerId, entryId};
}
- long[] positionPair = new long[]{Long.valueOf(ledgerIdAndEntryId[0]).longValue(),
- Long.valueOf(ledgerIdAndEntryId[1]).longValue()};
- publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, positionPair);
- break;
+ } catch (NumberFormatException e) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
}
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
}
}
+ return null;
}
- public MessageDupStatus isDuplicateReplV2(PublishContext publishContext, ByteBuf headersAndPayload) {
+ private long[] getReplSourcePosition(PublishContext publishContext) {
Object positionPairObj = publishContext.getProperty(MSG_PROP_REPL_SOURCE_POSITION);
- if (positionPairObj == null || !(positionPairObj instanceof long[])) {
- log.error()
- .attr("producerName", publishContext.getProducerName())
- .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
- .attr("sequenceId", publishContext.getSequenceId())
- .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
- .log("Message cannot determine whether it is duplicated due to the acquired message "
- + "props being invalid, prop not in expected format");
- return MessageDupStatus.Unknown;
+ if (!(positionPairObj instanceof long[]) || ((long[]) positionPairObj).length < 2) {
+ return null;
}
+ return (long[]) positionPairObj;
+ }
- long[] positionPair = (long[]) positionPairObj;
- long replSequenceLId = positionPair[0];
- long replSequenceEId = positionPair[1];
+ private void logInvalidReplSourcePosition(PublishContext publishContext, String message) {
+ log.error()
+ .attr("producerName", publishContext.getProducerName())
+ .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
+ .attr("sequenceId", publishContext.getSequenceId())
+ .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
+ .log(message);
+ }
- String lastSequenceLIdKey = publishContext.getProducerName() + "_LID";
- String lastSequenceEIdKey = publishContext.getProducerName() + "_EID";
- synchronized (highestSequencedPushed) {
- Long lastSequenceLIdPushed = highestSequencedPushed.get(lastSequenceLIdKey);
- Long lastSequenceEIdPushed = highestSequencedPushed.get(lastSequenceEIdKey);
- if (lastSequenceLIdPushed != null && lastSequenceEIdPushed != null
- && (replSequenceLId < lastSequenceLIdPushed.longValue()
- || (replSequenceLId == lastSequenceLIdPushed.longValue()
- && replSequenceEId <= lastSequenceEIdPushed.longValue()))) {
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPushed", lastSequenceLIdPushed)
- .attr("lastSequenceEIdPushed", lastSequenceEIdPushed)
- .log("Message identified as duplicated");
+ private void recoverReplWatermark(String producerName, long ledgerId, long entryId) {
+ ReplSourcePosition replSourcePosition = new ReplSourcePosition(ledgerId, entryId);
+ highestReplPositionPushed.merge(producerName, replSourcePosition, this::maxReplPosition);
+ highestReplPositionPersisted.merge(producerName, replSourcePosition, this::maxReplPosition);
+ }
- // Also need to check sequence ids that has been persisted.
- // If current message's seq id is smaller or equals to the
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted" than its definitely a dup
- // If current message's seq id is between "lastSequenceLIdPushed:lastSequenceEIdPushed" and
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted", then we cannot be sure whether the message
- // is a dup or not we should return an error to the producer for the latter case so that it can retry
- // at a future time
- Long lastSequenceLIdPersisted = highestSequencedPersisted.get(lastSequenceLIdKey);
- Long lastSequenceEIdPersisted = highestSequencedPersisted.get(lastSequenceEIdKey);
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPersisted", lastSequenceLIdPersisted)
- .attr("lastSequenceEIdPersisted", lastSequenceEIdPersisted)
- .log("Message identified as duplicated producer=. publishing:, latest" + "persisted");
- if (lastSequenceLIdPersisted != null && lastSequenceEIdPersisted != null
- && (replSequenceLId < lastSequenceLIdPersisted.longValue()
- || (replSequenceLId == lastSequenceLIdPersisted.longValue()
- && replSequenceEId <= lastSequenceEIdPersisted))) {
- return MessageDupStatus.Dup;
- } else {
- return MessageDupStatus.Unknown;
- }
+ private void recordReplWatermarkPersisted(String producerName, long ledgerId, long entryId) {
+ highestReplPositionPersisted.merge(producerName, new ReplSourcePosition(ledgerId, entryId),
+ this::maxReplPosition);
+ }
+
+ private boolean isReplPositionAtOrBefore(long ledgerId, long entryId, ReplSourcePosition watermark) {
+ return ledgerId < watermark.ledgerId()
+ || (ledgerId == watermark.ledgerId() && entryId <= watermark.entryId());
+ }
+
+ private ReplSourcePosition maxReplPosition(ReplSourcePosition current, ReplSourcePosition candidate) {
+ return isReplPositionAtOrBefore(candidate.ledgerId(), candidate.entryId(), current) ? current : candidate;
+ }
+
+ private void logUnexpectedReplSourcePosition(String producerName, String value) {
+ log.warn()
+ .attr("producerName", producerName)
+ .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
+ .attr("value", value)
+ .log("Unexpected");
+ }
+
+ private void logIncompleteReplSourcePositionSnapshot(String producerName, Long ledgerId, Long entryId) {
+ log.error()
+ .attr("producerName", producerName)
+ .attr("ledgerId", ledgerId)
+ .attr("entryId", entryId)
+ .log("Incomplete geo-replication deduplication snapshot");
+ }
+
+ private void recoverProducerSequenceIdFromMetadata(MessageMetadata md) {
+ final var producerName = md.getProducerName();
+ final var sequenceId = Math.max(md.getHighestSequenceId(), md.getSequenceId());
+ highestSequencedPushed.merge(producerName, sequenceId, Math::max);
Review Comment:
**[BUG] replay with merge(max) can resurrect the sequence ceiling of a purged producer**
Replay used to take the last entry per producer (`put`). It now keeps the maximum (`merge(..., Math::max)`).
That differs when a producer name is purged and then reused with a lower sequence id. `purgeInactiveProducers` removes the producer from both maps and snapshots at the current mark-delete position, so the old entries stay in the replay range. Example: producer `P` writes sequence 100, goes inactive and is purged, then reconnects and writes sequence 0. After an unload, replay sees 100 and then 0. With `merge(max)` the recovered ceiling is 100, so new sequences 1-100 from `P` are dropped as duplicates. Before this change the ceiling was 0.
The snapshot recovery and the `recordMessagePersisted*` paths do not need `max` for the V1 fix. Could replay keep `put` here, and for the replication watermark in `recoverReplWatermark`? Replay is in log order, so the last entry is the right one. If you want to keep `max`, a test for the purge-and-reuse case would show it is safe.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -262,24 +278,54 @@ private CompletableFuture<Void> replayCursor(ManagedCursor cursor) {
snapshotCounter = 0;
highestSequencedPushed.clear();
highestSequencedPersisted.clear();
+ highestReplPositionPushed.clear();
+ highestReplPositionPersisted.clear();
inactiveProducers.clear();
// Load the sequence ids from the snapshot in the cursor properties
+ Map<String, Long> replLedgerIds = new HashMap<>();
+ Map<String, Long> replEntryIds = new HashMap<>();
managedCursor.getProperties().forEach((k, v) -> {
- producerRemoved(k);
+ // Geo-replication V2 keys are source-position watermarks, not producer lifecycle state.
+ if (isReplSequenceKey(k)) {
+ String baseProducerName = getBaseProducerName(k);
+ if (k.endsWith(REPL_LEDGER_ID_SUFFIX)) {
+ replLedgerIds.put(baseProducerName, v);
+ } else {
+ replEntryIds.put(baseProducerName, v);
+ }
+ return;
+ }
highestSequencedPushed.put(k, v);
highestSequencedPersisted.put(k, v);
+ producerRemoved(k);
+ });
+ replLedgerIds.forEach((producerName, ledgerId) -> {
+ Long entryId = replEntryIds.get(producerName);
+ if (entryId != null) {
+ ReplSourcePosition position = new ReplSourcePosition(ledgerId, entryId);
+ highestReplPositionPushed.put(producerName, position);
+ highestReplPositionPersisted.put(producerName, position);
+ } else {
+ logIncompleteReplSourcePositionSnapshot(producerName, ledgerId, null);
+ }
+ });
+ replEntryIds.forEach((producerName, entryId) -> {
+ if (!replLedgerIds.containsKey(producerName)) {
+ logIncompleteReplSourcePositionSnapshot(producerName, null, entryId);
+ }
});
// Replay all the entries and apply all the sequence ids updates
log.info()
.attr("numberOfEntries", managedCursor.getNumberOfEntries())
.log("Replaying entries for deduplication");
return replayTask.replay(cursor, (__, buffer) -> {
final var metadata = Commands.parseMessageMetadata(buffer);
- final var producerName = metadata.getProducerName();
- final var sequenceId = Math.max(metadata.getHighestSequenceId(), metadata.getSequenceId());
- highestSequencedPushed.put(producerName, sequenceId);
- highestSequencedPersisted.put(producerName, sequenceId);
- producerRemoved(producerName);
+ if (metadata.hasMarkerType() || getReplProducerName(metadata) == null) {
+ recoverProducerSequenceIdFromMetadata(metadata);
+ } else {
+ // Rebuild replication watermarks from entries written after the last dedup snapshot.
+ recoverReplWatermarkFromMetadata(metadata);
Review Comment:
Thanks, this covers it. Replay now falls back to the original producer's sequence recovery when the entry has no valid source position, and `testReplayRecoversLegacyV1ReplicationSequenceIdAfterUnload` checks the unload/reload case with a V1 entry. I left one question on the new `merge(max)` in replay.
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#discussion_r3475936531
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -603,9 +687,17 @@ private CompletableFuture<Void> takeSnapshot(Position position) {
}
Map<String, Long> snapshot = new TreeMap<>();
+ highestReplPositionPersisted.forEach((producerName, replSourcePosition) -> {
+ snapshot.put(producerName + REPL_LEDGER_ID_SUFFIX, replSourcePosition.ledgerId());
+ snapshot.put(producerName + REPL_ENTRY_ID_SUFFIX, replSourcePosition.entryId());
+ });
+ int[] normalProducerCount = new int[1];
Review Comment:
Done. Changed this to a plain `int` with a normal loop.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -356,104 +413,141 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header
MessageMetadata md = Commands.parseMessageMetadata(headersAndPayload);
headersAndPayload.readerIndex(readerIndex);
- List<KeyValue> kvPairList = md.getPropertiesList();
- for (KeyValue kvPair : kvPairList) {
- if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
- if (!kvPair.getValue().contains(":")) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
- }
- String[] ledgerIdAndEntryId = kvPair.getValue().split(":");
- if (ledgerIdAndEntryId.length != 2 || !StringUtils.isNumeric(ledgerIdAndEntryId[0])
- || !StringUtils.isNumeric(ledgerIdAndEntryId[1])) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
+ ReplSourcePosition position = getReplSourcePosition(md, publishContext.getProducerName());
+ if (position != null) {
+ publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION,
+ new long[]{position.ledgerId(), position.entryId()});
+ }
+ }
+ }
+
+ private ReplSourcePosition getReplSourcePosition(MessageMetadata md, String producerName) {
+ List<KeyValue> kvPairList = md.getPropertiesList();
+ for (KeyValue kvPair : kvPairList) {
+ if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
+ String[] ledgerIdAndEntryId = kvPair.getValue().split(":", -1);
+ if (ledgerIdAndEntryId.length != 2) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
+ }
+ try {
+ long ledgerId = Long.parseLong(ledgerIdAndEntryId[0]);
+ long entryId = Long.parseLong(ledgerIdAndEntryId[1]);
+ if (ledgerId >= 0 && entryId >= 0) {
+ return new ReplSourcePosition(ledgerId, entryId);
}
- long[] positionPair = new long[]{Long.valueOf(ledgerIdAndEntryId[0]).longValue(),
- Long.valueOf(ledgerIdAndEntryId[1]).longValue()};
- publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, positionPair);
- break;
+ } catch (NumberFormatException e) {
+ // Log below.
Review Comment:
Done. The `NumberFormatException` branch now logs immediately and returns, avoiding the duplicate fall-through log.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -264,10 +299,30 @@ private CompletableFuture<Void> replayCursor(ManagedCursor cursor) {
highestSequencedPersisted.clear();
inactiveProducers.clear();
// Load the sequence ids from the snapshot in the cursor properties
+ Map<String, Long> replLedgerIds = new HashMap<>();
+ Map<String, Long> replEntryIds = new HashMap<>();
managedCursor.getProperties().forEach((k, v) -> {
- producerRemoved(k);
+ // Geo-replication V2 keys are source-position watermarks, not producer lifecycle state.
+ if (isReplSequenceKey(k)) {
+ String baseProducerName = getBaseProducerName(k);
+ if (k.endsWith(REPL_LEDGER_ID_SUFFIX)) {
+ replLedgerIds.put(baseProducerName, v);
+ } else {
+ replEntryIds.put(baseProducerName, v);
+ }
+ return;
+ }
highestSequencedPushed.put(k, v);
highestSequencedPersisted.put(k, v);
+ producerRemoved(k);
+ });
+ replLedgerIds.forEach((producerName, ledgerId) -> {
+ Long entryId = replEntryIds.get(producerName);
+ if (entryId != null) {
+ ReplSourcePosition position = new ReplSourcePosition(ledgerId, entryId);
+ highestReplPositionPushed.put(producerName, position);
+ highestReplPositionPersisted.put(producerName, position);
+ }
Review Comment:
Done. Added error logging for incomplete geo-replication watermark snapshot pairs when either `_LID` or `_EID` is missing.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -276,6 +331,8 @@ private CompletableFuture<Void> replayCursor(ManagedCursor cursor) {
return replayTask.replay(cursor, (__, buffer) -> {
final var metadata = Commands.parseMessageMetadata(buffer);
final var producerName = metadata.getProducerName();
+ // Rebuild replication watermarks from entries written after the last dedup snapshot.
+ recoverReplWatermarkFromMetadata(metadata);
final var sequenceId = Math.max(metadata.getHighestSequenceId(), metadata.getSequenceId());
Review Comment:
Done. Replay now restores marker/normal messages into `highestSequenced*` and replicated geo/shadow messages into `highestReplPosition*`.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -102,6 +103,28 @@ public MessageDupUnknownException(String topicName, String producerName) {
}
}
+ @VisibleForTesting
+ record ReplSourcePosition(long ledgerId, long entryId) implements Comparable<ReplSourcePosition> {
+ @Override
+ public int compareTo(ReplSourcePosition other) {
+ int ledgerComparison = Long.compare(ledgerId, other.ledgerId);
+ if (ledgerComparison != 0) {
+ return ledgerComparison;
+ }
+ return Long.compare(entryId, other.entryId);
+ }
+
+ static ReplSourcePosition max(ReplSourcePosition first, ReplSourcePosition second) {
+ return first.compareTo(second) >= 0 ? first : second;
+ }
+ }
+
+ private static class ReplDedupCheck {
Review Comment:
Done. Removed `ReplDedupCheck` and replaced it with method-local variables in `isDuplicateReplV2`.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -356,104 +413,141 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header
MessageMetadata md = Commands.parseMessageMetadata(headersAndPayload);
headersAndPayload.readerIndex(readerIndex);
- List<KeyValue> kvPairList = md.getPropertiesList();
- for (KeyValue kvPair : kvPairList) {
- if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
- if (!kvPair.getValue().contains(":")) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
- }
- String[] ledgerIdAndEntryId = kvPair.getValue().split(":");
- if (ledgerIdAndEntryId.length != 2 || !StringUtils.isNumeric(ledgerIdAndEntryId[0])
- || !StringUtils.isNumeric(ledgerIdAndEntryId[1])) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
+ ReplSourcePosition position = getReplSourcePosition(md, publishContext.getProducerName());
+ if (position != null) {
+ publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION,
+ new long[]{position.ledgerId(), position.entryId()});
+ }
+ }
+ }
+
+ private ReplSourcePosition getReplSourcePosition(MessageMetadata md, String producerName) {
+ List<KeyValue> kvPairList = md.getPropertiesList();
+ for (KeyValue kvPair : kvPairList) {
+ if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
+ String[] ledgerIdAndEntryId = kvPair.getValue().split(":", -1);
+ if (ledgerIdAndEntryId.length != 2) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
+ }
+ try {
+ long ledgerId = Long.parseLong(ledgerIdAndEntryId[0]);
+ long entryId = Long.parseLong(ledgerIdAndEntryId[1]);
+ if (ledgerId >= 0 && entryId >= 0) {
+ return new ReplSourcePosition(ledgerId, entryId);
}
- long[] positionPair = new long[]{Long.valueOf(ledgerIdAndEntryId[0]).longValue(),
- Long.valueOf(ledgerIdAndEntryId[1]).longValue()};
- publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, positionPair);
- break;
+ } catch (NumberFormatException e) {
+ // Log below.
}
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
}
}
+ return null;
}
- public MessageDupStatus isDuplicateReplV2(PublishContext publishContext, ByteBuf headersAndPayload) {
+ private ReplSourcePosition getReplSourcePosition(PublishContext publishContext) {
Object positionPairObj = publishContext.getProperty(MSG_PROP_REPL_SOURCE_POSITION);
- if (positionPairObj == null || !(positionPairObj instanceof long[])) {
- log.error()
- .attr("producerName", publishContext.getProducerName())
- .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
- .attr("sequenceId", publishContext.getSequenceId())
- .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
- .log("Message cannot determine whether it is duplicated due to the acquired message "
- + "props being invalid, prop not in expected format");
- return MessageDupStatus.Unknown;
+ if (!(positionPairObj instanceof long[]) || ((long[]) positionPairObj).length < 2) {
+ return null;
}
-
long[] positionPair = (long[]) positionPairObj;
- long replSequenceLId = positionPair[0];
- long replSequenceEId = positionPair[1];
+ return new ReplSourcePosition(positionPair[0], positionPair[1]);
+ }
- String lastSequenceLIdKey = publishContext.getProducerName() + "_LID";
- String lastSequenceEIdKey = publishContext.getProducerName() + "_EID";
- synchronized (highestSequencedPushed) {
- Long lastSequenceLIdPushed = highestSequencedPushed.get(lastSequenceLIdKey);
- Long lastSequenceEIdPushed = highestSequencedPushed.get(lastSequenceEIdKey);
- if (lastSequenceLIdPushed != null && lastSequenceEIdPushed != null
- && (replSequenceLId < lastSequenceLIdPushed.longValue()
- || (replSequenceLId == lastSequenceLIdPushed.longValue()
- && replSequenceEId <= lastSequenceEIdPushed.longValue()))) {
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPushed", lastSequenceLIdPushed)
- .attr("lastSequenceEIdPushed", lastSequenceEIdPushed)
- .log("Message identified as duplicated");
+ private void logInvalidReplSourcePosition(PublishContext publishContext, String message) {
+ log.error()
+ .attr("producerName", publishContext.getProducerName())
+ .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
+ .attr("sequenceId", publishContext.getSequenceId())
+ .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
+ .log(message);
+ }
- // Also need to check sequence ids that has been persisted.
- // If current message's seq id is smaller or equals to the
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted" than its definitely a dup
- // If current message's seq id is between "lastSequenceLIdPushed:lastSequenceEIdPushed" and
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted", then we cannot be sure whether the message
- // is a dup or not we should return an error to the producer for the latter case so that it can retry
- // at a future time
- Long lastSequenceLIdPersisted = highestSequencedPersisted.get(lastSequenceLIdKey);
- Long lastSequenceEIdPersisted = highestSequencedPersisted.get(lastSequenceEIdKey);
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPersisted", lastSequenceLIdPersisted)
- .attr("lastSequenceEIdPersisted", lastSequenceEIdPersisted)
- .log("Message identified as duplicated producer=. publishing:, latest" + "persisted");
- if (lastSequenceLIdPersisted != null && lastSequenceEIdPersisted != null
- && (replSequenceLId < lastSequenceLIdPersisted.longValue()
- || (replSequenceLId == lastSequenceLIdPersisted.longValue()
- && replSequenceEId <= lastSequenceEIdPersisted))) {
- return MessageDupStatus.Dup;
- } else {
- return MessageDupStatus.Unknown;
- }
+ private void recoverReplWatermark(String producerName, ReplSourcePosition replSourcePosition) {
+ highestReplPositionPushed.merge(producerName, replSourcePosition, ReplSourcePosition::max);
Review Comment:
Done. Removed the max/ordering merge logic and now stores the recovered watermark directly.
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#issuecomment-4539428315
all CI pass in my local repo: https://github.com/void-ptr974/pulsar/actions/runs/26407472425
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#issuecomment-4701903833
Hi @poorbarcode, this PR touches geo-replication V2 dedup watermark recovery and cleanup behavior. Since it is close to the geo-replicator lifecycle work you have been looking at, could you help take a look when you have time? Thanks!
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#discussion_r3475939122
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java:
##########
@@ -272,6 +273,107 @@ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
admin2.topics().unload(topicName);
}
+ @Test
+ public void testGeoReplDedupAfterTargetUnloadBeforeSnapshot() throws Exception {
+ // Drop send receipts so the source replication cursor cannot advance while target messages are persisted.
+ AtomicBoolean stuckSendReceipt = new AtomicBoolean(true);
+ Runnable cleanInjection = injectReplicatorClientCnx((conf, eventLoopGroup) -> new ClientCnx(
+ InstrumentProvider.NOOP, conf, eventLoopGroup) {
+ @Override
+ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
+ if (!stuckSendReceipt.get()) {
+ super.handleSendReceipt(sendReceipt);
+ }
+ }
+ });
+
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + nonReplicatedNamespace + "/tp_");
+ final String subscription = "s1";
+ Producer<Integer> producer = null;
+ Consumer<Integer> consumer = null;
+ try {
+ admin1.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createSubscription(topicName, subscription, MessageId.earliest);
+
+ PersistentTopic persistentTopic2 =
+ (PersistentTopic) pulsar2.getBrokerService().getTopic(topicName, false).join().get();
+ admin2.topicPolicies().setDeduplicationStatus(topicName, true);
+ Awaitility.await().untilAsserted(() -> {
+ MessageDeduplication messageDeduplication2 = persistentTopic2.getMessageDeduplication();
+ assertEquals(String.valueOf(messageDeduplication2.getStatus()), "Enabled");
+ });
+
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1, cluster2));
+ waitReplicatorStarted(topicName);
+ PersistentTopic persistentTopic1 =
+ (PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
+
+ producer = client1.newProducer(Schema.INT32).topic(topicName).create();
+ int messageCount = 4;
+ for (int i = 0; i < messageCount; i++) {
+ producer.send(i);
+ }
+
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertEquals(admin2.topics().getStats(topicName).getMsgInCounter(), messageCount);
+ });
+
+ GeoPersistentReplicator replicator =
+ (GeoPersistentReplicator) persistentTopic1.getReplicators().get(cluster2);
+ assertNotNull(replicator);
+ AtomicReference<ClientCnx> replicatorClientCnx = new AtomicReference<>();
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertNotNull(replicator.producer);
+ ClientCnx clientCnx = replicator.producer.getClientCnx();
+ assertNotNull(clientCnx);
+ replicatorClientCnx.set(clientCnx);
+ });
+
+ // Reload target before a dedup snapshot has to cover the latest replicated source positions.
+ admin2.topics().unload(topicName);
+ pulsar2.getBrokerService().getTopic(topicName, false).join();
+
+ // Reconnect the source replicator. It replays from the old replication cursor and target must deduplicate.
+ stuckSendReceipt.set(false);
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertNotNull(replicator.producer);
+ ClientCnx clientCnx = replicator.producer.getClientCnx();
+ assertNotNull(clientCnx);
+ replicatorClientCnx.set(clientCnx);
+ });
+ replicatorClientCnx.get().ctx().channel().close();
+
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertEquals(replicator.getCursor().getNumberOfEntriesInBacklog(true), 0);
Review Comment:
Done. The backlog check now re-fetches the current source topic and replicator before asserting.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java:
##########
@@ -272,6 +273,107 @@ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
admin2.topics().unload(topicName);
}
+ @Test
+ public void testGeoReplDedupAfterTargetUnloadBeforeSnapshot() throws Exception {
+ // Drop send receipts so the source replication cursor cannot advance while target messages are persisted.
+ AtomicBoolean stuckSendReceipt = new AtomicBoolean(true);
+ Runnable cleanInjection = injectReplicatorClientCnx((conf, eventLoopGroup) -> new ClientCnx(
+ InstrumentProvider.NOOP, conf, eventLoopGroup) {
+ @Override
+ protected void handleSendReceipt(CommandSendReceipt sendReceipt) {
+ if (!stuckSendReceipt.get()) {
+ super.handleSendReceipt(sendReceipt);
+ }
+ }
+ });
+
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://" + nonReplicatedNamespace + "/tp_");
+ final String subscription = "s1";
+ Producer<Integer> producer = null;
+ Consumer<Integer> consumer = null;
+ try {
+ admin1.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createNonPartitionedTopic(topicName);
+ admin2.topics().createSubscription(topicName, subscription, MessageId.earliest);
+
+ PersistentTopic persistentTopic2 =
+ (PersistentTopic) pulsar2.getBrokerService().getTopic(topicName, false).join().get();
+ admin2.topicPolicies().setDeduplicationStatus(topicName, true);
+ Awaitility.await().untilAsserted(() -> {
+ MessageDeduplication messageDeduplication2 = persistentTopic2.getMessageDeduplication();
+ assertEquals(String.valueOf(messageDeduplication2.getStatus()), "Enabled");
+ });
+
+ admin1.topics().setReplicationClusters(topicName, Arrays.asList(cluster1, cluster2));
+ waitReplicatorStarted(topicName);
+ PersistentTopic persistentTopic1 =
+ (PersistentTopic) pulsar1.getBrokerService().getTopic(topicName, false).join().get();
+
+ producer = client1.newProducer(Schema.INT32).topic(topicName).create();
+ int messageCount = 4;
+ for (int i = 0; i < messageCount; i++) {
+ producer.send(i);
+ }
+
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertEquals(admin2.topics().getStats(topicName).getMsgInCounter(), messageCount);
+ });
+
+ GeoPersistentReplicator replicator =
+ (GeoPersistentReplicator) persistentTopic1.getReplicators().get(cluster2);
+ assertNotNull(replicator);
+ AtomicReference<ClientCnx> replicatorClientCnx = new AtomicReference<>();
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertNotNull(replicator.producer);
+ ClientCnx clientCnx = replicator.producer.getClientCnx();
+ assertNotNull(clientCnx);
+ replicatorClientCnx.set(clientCnx);
+ });
+
+ // Reload target before a dedup snapshot has to cover the latest replicated source positions.
+ admin2.topics().unload(topicName);
+ pulsar2.getBrokerService().getTopic(topicName, false).join();
+
+ // Reconnect the source replicator. It replays from the old replication cursor and target must deduplicate.
+ stuckSendReceipt.set(false);
+ Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
+ assertNotNull(replicator.producer);
Review Comment:
Done. The test now re-fetches the current source topic and replicator after reload before checking the connection.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -356,104 +413,141 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header
MessageMetadata md = Commands.parseMessageMetadata(headersAndPayload);
headersAndPayload.readerIndex(readerIndex);
- List<KeyValue> kvPairList = md.getPropertiesList();
- for (KeyValue kvPair : kvPairList) {
- if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
- if (!kvPair.getValue().contains(":")) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
- }
- String[] ledgerIdAndEntryId = kvPair.getValue().split(":");
- if (ledgerIdAndEntryId.length != 2 || !StringUtils.isNumeric(ledgerIdAndEntryId[0])
- || !StringUtils.isNumeric(ledgerIdAndEntryId[1])) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
+ ReplSourcePosition position = getReplSourcePosition(md, publishContext.getProducerName());
+ if (position != null) {
+ publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION,
+ new long[]{position.ledgerId(), position.entryId()});
+ }
+ }
+ }
+
+ private ReplSourcePosition getReplSourcePosition(MessageMetadata md, String producerName) {
+ List<KeyValue> kvPairList = md.getPropertiesList();
+ for (KeyValue kvPair : kvPairList) {
+ if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
+ String[] ledgerIdAndEntryId = kvPair.getValue().split(":", -1);
+ if (ledgerIdAndEntryId.length != 2) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
+ }
+ try {
+ long ledgerId = Long.parseLong(ledgerIdAndEntryId[0]);
+ long entryId = Long.parseLong(ledgerIdAndEntryId[1]);
+ if (ledgerId >= 0 && entryId >= 0) {
+ return new ReplSourcePosition(ledgerId, entryId);
}
- long[] positionPair = new long[]{Long.valueOf(ledgerIdAndEntryId[0]).longValue(),
- Long.valueOf(ledgerIdAndEntryId[1]).longValue()};
- publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, positionPair);
- break;
+ } catch (NumberFormatException e) {
+ // Log below.
}
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
}
}
+ return null;
}
- public MessageDupStatus isDuplicateReplV2(PublishContext publishContext, ByteBuf headersAndPayload) {
+ private ReplSourcePosition getReplSourcePosition(PublishContext publishContext) {
Object positionPairObj = publishContext.getProperty(MSG_PROP_REPL_SOURCE_POSITION);
- if (positionPairObj == null || !(positionPairObj instanceof long[])) {
- log.error()
- .attr("producerName", publishContext.getProducerName())
- .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
- .attr("sequenceId", publishContext.getSequenceId())
- .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
- .log("Message cannot determine whether it is duplicated due to the acquired message "
- + "props being invalid, prop not in expected format");
- return MessageDupStatus.Unknown;
+ if (!(positionPairObj instanceof long[]) || ((long[]) positionPairObj).length < 2) {
+ return null;
}
-
long[] positionPair = (long[]) positionPairObj;
- long replSequenceLId = positionPair[0];
- long replSequenceEId = positionPair[1];
+ return new ReplSourcePosition(positionPair[0], positionPair[1]);
+ }
- String lastSequenceLIdKey = publishContext.getProducerName() + "_LID";
- String lastSequenceEIdKey = publishContext.getProducerName() + "_EID";
- synchronized (highestSequencedPushed) {
- Long lastSequenceLIdPushed = highestSequencedPushed.get(lastSequenceLIdKey);
- Long lastSequenceEIdPushed = highestSequencedPushed.get(lastSequenceEIdKey);
- if (lastSequenceLIdPushed != null && lastSequenceEIdPushed != null
- && (replSequenceLId < lastSequenceLIdPushed.longValue()
- || (replSequenceLId == lastSequenceLIdPushed.longValue()
- && replSequenceEId <= lastSequenceEIdPushed.longValue()))) {
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPushed", lastSequenceLIdPushed)
- .attr("lastSequenceEIdPushed", lastSequenceEIdPushed)
- .log("Message identified as duplicated");
+ private void logInvalidReplSourcePosition(PublishContext publishContext, String message) {
+ log.error()
+ .attr("producerName", publishContext.getProducerName())
+ .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
+ .attr("sequenceId", publishContext.getSequenceId())
+ .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
+ .log(message);
+ }
- // Also need to check sequence ids that has been persisted.
- // If current message's seq id is smaller or equals to the
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted" than its definitely a dup
- // If current message's seq id is between "lastSequenceLIdPushed:lastSequenceEIdPushed" and
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted", then we cannot be sure whether the message
- // is a dup or not we should return an error to the producer for the latter case so that it can retry
- // at a future time
- Long lastSequenceLIdPersisted = highestSequencedPersisted.get(lastSequenceLIdKey);
- Long lastSequenceEIdPersisted = highestSequencedPersisted.get(lastSequenceEIdKey);
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPersisted", lastSequenceLIdPersisted)
- .attr("lastSequenceEIdPersisted", lastSequenceEIdPersisted)
- .log("Message identified as duplicated producer=. publishing:, latest" + "persisted");
- if (lastSequenceLIdPersisted != null && lastSequenceEIdPersisted != null
- && (replSequenceLId < lastSequenceLIdPersisted.longValue()
- || (replSequenceLId == lastSequenceLIdPersisted.longValue()
- && replSequenceEId <= lastSequenceEIdPersisted))) {
- return MessageDupStatus.Dup;
- } else {
- return MessageDupStatus.Unknown;
- }
+ private void recoverReplWatermark(String producerName, ReplSourcePosition replSourcePosition) {
+ highestReplPositionPushed.merge(producerName, replSourcePosition, ReplSourcePosition::max);
+ highestReplPositionPersisted.merge(producerName, replSourcePosition, ReplSourcePosition::max);
+ }
+
+ private void recordReplWatermarkPersisted(String producerName, ReplSourcePosition replSourcePosition) {
+ highestReplPositionPersisted.merge(producerName, replSourcePosition, ReplSourcePosition::max);
+ }
+
+ private boolean isReplPositionAtOrBefore(ReplSourcePosition position, ReplSourcePosition watermark) {
+ return position.compareTo(watermark) <= 0;
+ }
+
+ private void logUnexpectedReplSourcePosition(String producerName, String value) {
+ log.warn()
+ .attr("producerName", producerName)
+ .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
+ .attr("value", value)
+ .log("Unexpected");
+ }
+
+ @VisibleForTesting
+ void recoverReplWatermarkFromMetadata(MessageMetadata md) {
+ if (md.hasMarkerType()) {
+ return;
+ }
+ String replProducerName = getReplProducerName(md);
+ if (replProducerName != null) {
+ ReplSourcePosition replSourcePosition = getReplSourcePosition(md, replProducerName);
+ if (replSourcePosition != null) {
+ recoverReplWatermark(replProducerName, replSourcePosition);
}
- highestSequencedPushed.put(lastSequenceLIdKey, replSequenceLId);
- highestSequencedPushed.put(lastSequenceEIdKey, replSequenceEId);
}
+ }
+
+ @VisibleForTesting
+ String getReplProducerName(MessageMetadata md) {
+ final var shadowSourceTopic = topic.getShadowSourceTopic();
+ if (md.hasReplicatedFrom() && shadowSourceTopic.isPresent()) {
+ return ShadowReplicator.getShadowProducerName(replicatorPrefix,
+ shadowSourceTopic.get().toString(), topic.getName());
+ }
+ if (md.hasReplicatedFrom()) {
+ return AbstractReplicator.getReplicatorName(replicatorPrefix, md.getReplicatedFrom())
+ + AbstractReplicator.REPL_PRODUCER_NAME_DELIMITER
+ + pulsar.getConfiguration().getClusterName();
+ }
+ if (Producer.isRemoteOrShadow(md.getProducerName(), replicatorPrefix)) {
+ return md.getProducerName();
+ }
+ return null;
+ }
+
+ public MessageDupStatus isDuplicateReplV2(PublishContext publishContext, ByteBuf headersAndPayload) {
+ ReplSourcePosition replSourcePosition = getReplSourcePosition(publishContext);
+ if (replSourcePosition == null) {
+ logInvalidReplSourcePosition(publishContext, "Message cannot determine whether it is duplicated due to "
+ + "the acquired message props being invalid, prop not in expected format");
+ return MessageDupStatus.Unknown;
+ }
+
+ String producerName = publishContext.getProducerName();
+ ReplDedupCheck check = new ReplDedupCheck();
+ highestReplPositionPushed.compute(producerName, (__, lastPushed) -> {
Review Comment:
Done. The in-memory watermark cache now uses the base producer name, while snapshot keys keep the `_LID` / `_EID` suffixes.
--
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] [fix][broker] Prevent duplicate geo-replicated messages after target topic reload [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #25860:
URL: https://github.com/apache/pulsar/pull/25860#discussion_r3475937981
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -356,104 +413,141 @@ private void setContextPropsIfRepl(PublishContext publishContext, ByteBuf header
MessageMetadata md = Commands.parseMessageMetadata(headersAndPayload);
headersAndPayload.readerIndex(readerIndex);
- List<KeyValue> kvPairList = md.getPropertiesList();
- for (KeyValue kvPair : kvPairList) {
- if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
- if (!kvPair.getValue().contains(":")) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
- }
- String[] ledgerIdAndEntryId = kvPair.getValue().split(":");
- if (ledgerIdAndEntryId.length != 2 || !StringUtils.isNumeric(ledgerIdAndEntryId[0])
- || !StringUtils.isNumeric(ledgerIdAndEntryId[1])) {
- log.warn()
- .attr("producerName", publishContext.getProducerName())
- .attr("MSG_PROP_REPL_SOURCE_POSITION", MSG_PROP_REPL_SOURCE_POSITION)
- .attr("value", kvPair.getValue())
- .log("Unexpected");
- break;
+ ReplSourcePosition position = getReplSourcePosition(md, publishContext.getProducerName());
+ if (position != null) {
+ publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION,
+ new long[]{position.ledgerId(), position.entryId()});
+ }
+ }
+ }
+
+ private ReplSourcePosition getReplSourcePosition(MessageMetadata md, String producerName) {
+ List<KeyValue> kvPairList = md.getPropertiesList();
+ for (KeyValue kvPair : kvPairList) {
+ if (kvPair.getKey().equals(MSG_PROP_REPL_SOURCE_POSITION)) {
+ String[] ledgerIdAndEntryId = kvPair.getValue().split(":", -1);
+ if (ledgerIdAndEntryId.length != 2) {
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
+ }
+ try {
+ long ledgerId = Long.parseLong(ledgerIdAndEntryId[0]);
+ long entryId = Long.parseLong(ledgerIdAndEntryId[1]);
+ if (ledgerId >= 0 && entryId >= 0) {
+ return new ReplSourcePosition(ledgerId, entryId);
}
- long[] positionPair = new long[]{Long.valueOf(ledgerIdAndEntryId[0]).longValue(),
- Long.valueOf(ledgerIdAndEntryId[1]).longValue()};
- publishContext.setProperty(MSG_PROP_REPL_SOURCE_POSITION, positionPair);
- break;
+ } catch (NumberFormatException e) {
+ // Log below.
}
+ logUnexpectedReplSourcePosition(producerName, kvPair.getValue());
+ return null;
}
}
+ return null;
}
- public MessageDupStatus isDuplicateReplV2(PublishContext publishContext, ByteBuf headersAndPayload) {
+ private ReplSourcePosition getReplSourcePosition(PublishContext publishContext) {
Object positionPairObj = publishContext.getProperty(MSG_PROP_REPL_SOURCE_POSITION);
- if (positionPairObj == null || !(positionPairObj instanceof long[])) {
- log.error()
- .attr("producerName", publishContext.getProducerName())
- .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
- .attr("sequenceId", publishContext.getSequenceId())
- .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
- .log("Message cannot determine whether it is duplicated due to the acquired message "
- + "props being invalid, prop not in expected format");
- return MessageDupStatus.Unknown;
+ if (!(positionPairObj instanceof long[]) || ((long[]) positionPairObj).length < 2) {
+ return null;
}
-
long[] positionPair = (long[]) positionPairObj;
- long replSequenceLId = positionPair[0];
- long replSequenceEId = positionPair[1];
+ return new ReplSourcePosition(positionPair[0], positionPair[1]);
+ }
- String lastSequenceLIdKey = publishContext.getProducerName() + "_LID";
- String lastSequenceEIdKey = publishContext.getProducerName() + "_EID";
- synchronized (highestSequencedPushed) {
- Long lastSequenceLIdPushed = highestSequencedPushed.get(lastSequenceLIdKey);
- Long lastSequenceEIdPushed = highestSequencedPushed.get(lastSequenceEIdKey);
- if (lastSequenceLIdPushed != null && lastSequenceEIdPushed != null
- && (replSequenceLId < lastSequenceLIdPushed.longValue()
- || (replSequenceLId == lastSequenceLIdPushed.longValue()
- && replSequenceEId <= lastSequenceEIdPushed.longValue()))) {
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPushed", lastSequenceLIdPushed)
- .attr("lastSequenceEIdPushed", lastSequenceEIdPushed)
- .log("Message identified as duplicated");
+ private void logInvalidReplSourcePosition(PublishContext publishContext, String message) {
+ log.error()
+ .attr("producerName", publishContext.getProducerName())
+ .attr("supportsReplDedupByLidAndEid", publishContext.supportsReplDedupByLidAndEid())
+ .attr("sequenceId", publishContext.getSequenceId())
+ .attr("propKey", MSG_PROP_REPL_SOURCE_POSITION)
+ .log(message);
+ }
- // Also need to check sequence ids that has been persisted.
- // If current message's seq id is smaller or equals to the
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted" than its definitely a dup
- // If current message's seq id is between "lastSequenceLIdPushed:lastSequenceEIdPushed" and
- // "lastSequenceLIdPersisted:lastSequenceEIdPersisted", then we cannot be sure whether the message
- // is a dup or not we should return an error to the producer for the latter case so that it can retry
- // at a future time
- Long lastSequenceLIdPersisted = highestSequencedPersisted.get(lastSequenceLIdKey);
- Long lastSequenceEIdPersisted = highestSequencedPersisted.get(lastSequenceEIdKey);
- log.debug()
- .attr("producerName", publishContext.getProducerName())
- .attr("replSequenceLId", replSequenceLId)
- .attr("replSequenceEId", replSequenceEId)
- .attr("lastSequenceLIdPersisted", lastSequenceLIdPersisted)
- .attr("lastSequenceEIdPersisted", lastSequenceEIdPersisted)
- .log("Message identified as duplicated producer=. publishing:, latest" + "persisted");
- if (lastSequenceLIdPersisted != null && lastSequenceEIdPersisted != null
- && (replSequenceLId < lastSequenceLIdPersisted.longValue()
- || (replSequenceLId == lastSequenceLIdPersisted.longValue()
- && replSequenceEId <= lastSequenceEIdPersisted))) {
- return MessageDupStatus.Dup;
- } else {
- return MessageDupStatus.Unknown;
- }
+ private void recoverReplWatermark(String producerName, ReplSourcePosition replSourcePosition) {
+ highestReplPositionPushed.merge(producerName, replSourcePosition, ReplSourcePosition::max);
+ highestReplPositionPersisted.merge(producerName, replSourcePosition, ReplSourcePosition::max);
+ }
+
+ private void recordReplWatermarkPersisted(String producerName, ReplSourcePosition replSourcePosition) {
+ highestReplPositionPersisted.merge(producerName, replSourcePosition, ReplSourcePosition::max);
Review Comment:
Done. Removed the watermark ordering/max merge path and now records the recovered watermark directly.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java:
##########
@@ -102,6 +103,28 @@ public MessageDupUnknownException(String topicName, String producerName) {
}
}
+ @VisibleForTesting
+ record ReplSourcePosition(long ledgerId, long entryId) implements Comparable<ReplSourcePosition> {
Review Comment:
Adjusted. The hot path now uses `long[]` for parsed source positions; `ReplSourcePosition` remains only as the map value for the persisted watermark pair.
--
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