You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zh...@apache.org on 2023/08/15 06:21:39 UTC
[pulsar] 03/03: [fix][broker] Fix message loss during topic compaction (#20980)
This is an automated email from the ASF dual-hosted git repository.
zhaocong pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8e88fbf1c5060bd643616c234a2e3c8320ea5d18
Author: coderzc <zh...@apache.org>
AuthorDate: Tue Aug 15 14:21:22 2023 +0800
[fix][broker] Fix message loss during topic compaction (#20980)
(cherry picked from commit 3ab420cd81c31ebd16213e14580d9e317bc0698d)
---
.../java/org/apache/pulsar/client/impl/RawBatchConverter.java | 10 ++++++++--
.../java/org/apache/pulsar/compaction/TwoPhaseCompactor.java | 2 +-
2 files changed, 9 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchConverter.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchConverter.java
index 5f476df68af..167cc1b699c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchConverter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawBatchConverter.java
@@ -52,6 +52,12 @@ public class RawBatchConverter {
}
public static List<ImmutableTriple<MessageId, String, Integer>> extractIdsAndKeysAndSize(RawMessage msg)
+ throws IOException {
+ return extractIdsAndKeysAndSize(msg, true);
+ }
+
+ public static List<ImmutableTriple<MessageId, String, Integer>> extractIdsAndKeysAndSize(RawMessage msg,
+ boolean extractNullKey)
throws IOException {
checkArgument(msg.getMessageIdData().getBatchIndex() == -1);
@@ -75,9 +81,9 @@ public class RawBatchConverter {
msg.getMessageIdData().getEntryId(),
msg.getMessageIdData().getPartition(),
i);
- if (!smm.isCompactedOut() && smm.hasPartitionKey()) {
+ if (!smm.isCompactedOut() && (extractNullKey || smm.hasPartitionKey())) {
idsAndKeysAndSize.add(ImmutableTriple.of(id,
- smm.getPartitionKey(),
+ smm.hasPartitionKey() ? smm.getPartitionKey() : null,
smm.hasPayloadSize() ? smm.getPayloadSize() : 0));
}
singleMessagePayload.release();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
index 29225253da1..d350ea08532 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
@@ -406,7 +406,7 @@ public class TwoPhaseCompactor extends Compactor {
protected List<ImmutableTriple<MessageId, String, Integer>> extractIdsAndKeysAndSizeFromBatch(RawMessage msg)
throws IOException {
- return RawBatchConverter.extractIdsAndKeysAndSize(msg);
+ return RawBatchConverter.extractIdsAndKeysAndSize(msg, false);
}
protected Optional<RawMessage> rebatchMessage(RawMessage msg, BiPredicate<String, MessageId> filter)