Fixed retention of keys in compaction #11287

merlimat · 2021-07-12T16:48:56Z

Motivation

This change fixes few issues in the compaction mechanism, the

When a reader is created, reading from "earliest" message, it should read the compacted data and then continue from the next message.
When the compaction consumer starts, it shouldn't seek to the beginning. This causes 2 issues:
- Rescanning of the topic each time the compaction runs
- Keys that are being dropped from the topic are also getting dropped from the compacted view, while in fact they should be there until explicitly deleted (with an empty message for a key).

The main source of the problem is that when creating a cursor on "earliest" message, the cursor gets automatically adjusted on the earliest message available to read. This confuses the check for the read-compacted because it may think the reader/consumer is already ahead of the compaction horizon.

Modifications

Introduced a "isFirstRead" flag to make sure we double check the start message id and use MessageId.earliest instead of the earliest available message to read on the topic. After the first read, the positioning will be fine.

Anonymitaet · 2021-07-13T01:56:11Z

thanks for your contribution. For this PR, do we need to update docs?

codelipenghui

The change looks good to me, I just left some minor comments about the test.

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionRetentionTest.java

codelipenghui · 2021-07-13T02:54:53Z

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionRetentionTest.java

+     * Compaction should retain expired keys in the compacted view
+     */
+    @Test
+    public void testCompaction() throws Exception {


For the test, I think we need to verify the raw data of the topic has been deleted. because the issue only happens when the raw data has been deleted.

codelipenghui · 2021-07-13T02:57:25Z

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionRetentionTest.java

+
+            assertTrue(expectedKeys.contains(msg.getKey()), "Unexpected key: " + msg.getKey());
+
+            System.err.println("Received: " + msg.getKey());


Is there any special reason here for using the System.err?

eolivelli

LGTM

This change fixes few issues in the compaction mechanism, the * When a reader is created, reading from "earliest" message, it should read the compacted data and then continue from the next message. * When the compaction consumer starts, it shouldn't seek to the beginning. This causes 2 issues: * Rescanning of the topic each time the compaction runs * Keys that are being dropped from the topic are also getting dropped from the compacted view, while in fact they should be there until explicitly deleted (with an empty message for a key). The main source of the problem is that when creating a cursor on "earliest" message, the cursor gets automatically adjusted on the earliest message available to read. This confuses the check for the read-compacted because it may think the reader/consumer is already ahead of the compaction horizon. Introduced a "isFirstRead" flag to make sure we double check the start message id and use `MessageId.earliest` instead of the earliest available message to read on the topic. After the first read, the positioning will be fine. (cherry picked from commit feb4ff1)

…branch-2.7.

…branch-2.7 apache#11384

…-2.7 #11384 (#11384)

### Motivation To fix the reader skipping remaining compacted data while the topic been unloaded. apache#11287 fixed the data skipped issue while the reader first time to read the messages with the earliest position. But if the reader has consumed some messages from the compacted ledger but not all, the start position will not be `earliest`, the broker will rewind the cursor for the reader to the next valid position of the original topic. So the remaining messages in the compacted ledger will be skipped. Here are the logs from broker: ``` 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.pulsar.broker.service.BrokerService - Created topic persistent://xxx/product-full-prod/5126 - dedup is disabled 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://xxx/product-full-prod/5126][xxx] Creating non-durable subscription at msg id 181759:14:-1:-1 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.NonDurableCursorImpl - [xxx/product-full-prod/persistent/5126] Created non-durable cursor read-position=221199:0 mark-delete-position=181759:13 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [xxx/product-full-prod/persistent/5126] Opened new cursor: NonDurableCursorImpl{ledger=xxx/product-full-prod/persistent/5126, ackPos=181759:13, readPos=221199:0} 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [xxx/product-full-prod/persistent/5126-xxx] Rewind from 221199:0 to 221199:0 ``` There some many compacted messages after `181759:13`, but the broker will not dispatch them to the reader. The issue also can be reproduced by the unit test that added in this PR. ### Modification If the cursor with `readCompacted = true`, just rewind to the next message of the mark delete position, so that the reader can continue to read the data from the compacted ledger. ### Verification New test added for testing the reader can get all the compacted messages and non-compacted message from the topic during the topic unloading.

…g. (#13629) ### Motivation To fix the reader skipping remaining compacted data while the topic has been unloaded. #11287 fixed the data skipped issue while the reader first time to read the messages with the earliest position. But if the reader has consumed some messages from the compacted ledger but not all, the start position will not be `earliest`, the broker will rewind the cursor for the reader to the next valid position of the original topic. So the remaining messages in the compacted ledger will be skipped. Here are the logs from the broker: ``` 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.pulsar.broker.service.BrokerService - Created topic persistent://xxx/product-full-prod/5126 - dedup is disabled 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://xxx/product-full-prod/5126][xxx] Creating non-durable subscription at msg id 181759:14:-1:-1 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.NonDurableCursorImpl - [xxx/product-full-prod/persistent/5126] Created non-durable cursor read-position=221199:0 mark-delete-position=181759:13 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [xxx/product-full-prod/persistent/5126] Opened new cursor: NonDurableCursorImpl{ledger=xxx/product-full-prod/persistent/5126, ackPos=181759:13, readPos=221199:0} 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [xxx/product-full-prod/persistent/5126-xxx] Rewind from 221199:0 to 221199:0 ``` There some many compacted messages after `181759:13`, but the broker will not dispatch them to the reader. The issue also can be reproduced by the unit test that was added in this PR. ### Modification If the cursor with `readCompacted = true`, just rewind to the next message of the mark delete position, so that the reader can continue to read the data from the compacted ledger. ### Verification A new test added for testing the reader can get all the compacted messages and non-compacted messages from the topic during the topic unloading.

…g. (#13629) ### Motivation To fix the reader skipping remaining compacted data while the topic has been unloaded. #11287 fixed the data skipped issue while the reader first time to read the messages with the earliest position. But if the reader has consumed some messages from the compacted ledger but not all, the start position will not be `earliest`, the broker will rewind the cursor for the reader to the next valid position of the original topic. So the remaining messages in the compacted ledger will be skipped. Here are the logs from the broker: ``` 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.pulsar.broker.service.BrokerService - Created topic persistent://xxx/product-full-prod/5126 - dedup is disabled 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://xxx/product-full-prod/5126][xxx] Creating non-durable subscription at msg id 181759:14:-1:-1 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.NonDurableCursorImpl - [xxx/product-full-prod/persistent/5126] Created non-durable cursor read-position=221199:0 mark-delete-position=181759:13 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [xxx/product-full-prod/persistent/5126] Opened new cursor: NonDurableCursorImpl{ledger=xxx/product-full-prod/persistent/5126, ackPos=181759:13, readPos=221199:0} 10:44:36.035 [bookkeeper-ml-scheduler-OrderedScheduler-4-0] INFO org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [xxx/product-full-prod/persistent/5126-xxx] Rewind from 221199:0 to 221199:0 ``` There some many compacted messages after `181759:13`, but the broker will not dispatch them to the reader. The issue also can be reproduced by the unit test that was added in this PR. ### Modification If the cursor with `readCompacted = true`, just rewind to the next message of the mark delete position, so that the reader can continue to read the data from the compacted ledger. ### Verification A new test added for testing the reader can get all the compacted messages and non-compacted messages from the topic during the topic unloading. (cherry picked from commit 07f131f)

This change fixes few issues in the compaction mechanism, the * When a reader is created, reading from "earliest" message, it should read the compacted data and then continue from the next message. * When the compaction consumer starts, it shouldn't seek to the beginning. This causes 2 issues: * Rescanning of the topic each time the compaction runs * Keys that are being dropped from the topic are also getting dropped from the compacted view, while in fact they should be there until explicitly deleted (with an empty message for a key). The main source of the problem is that when creating a cursor on "earliest" message, the cursor gets automatically adjusted on the earliest message available to read. This confuses the check for the read-compacted because it may think the reader/consumer is already ahead of the compaction horizon. Introduced a "isFirstRead" flag to make sure we double check the start message id and use `MessageId.earliest` instead of the earliest available message to read on the topic. After the first read, the positioning will be fine. (cherry picked from commit feb4ff1) (cherry picked from commit 9c39726)

### Motivation This change fixes few issues in the compaction mechanism, the * When a reader is created, reading from "earliest" message, it should read the compacted data and then continue from the next message. * When the compaction consumer starts, it shouldn't seek to the beginning. This causes 2 issues: * Rescanning of the topic each time the compaction runs * Keys that are being dropped from the topic are also getting dropped from the compacted view, while in fact they should be there until explicitly deleted (with an empty message for a key). The main source of the problem is that when creating a cursor on "earliest" message, the cursor gets automatically adjusted on the earliest message available to read. This confuses the check for the read-compacted because it may think the reader/consumer is already ahead of the compaction horizon. ### Modifications Introduced a "isFirstRead" flag to make sure we double check the start message id and use `MessageId.earliest` instead of the earliest available message to read on the topic. After the first read, the positioning will be fine.

Fixed retention of keys in compaction

abbb020

merlimat added type/bug The PR fixed a bug or issue reported a bug release/2.8.1 labels Jul 12, 2021

merlimat added this to the 2.9.0 milestone Jul 12, 2021

merlimat requested review from rdhabalia, jerrypeng, codelipenghui, massakam and congbobo184 July 12, 2021 16:48

merlimat self-assigned this Jul 12, 2021

codelipenghui added the doc-not-needed Your PR changes do not impact docs label Jul 13, 2021

codelipenghui requested review from 315157973 and hangc0276 July 13, 2021 02:43

codelipenghui reviewed Jul 13, 2021

View reviewed changes

Addressed comments

63c26b5

eolivelli approved these changes Jul 13, 2021

View reviewed changes

codelipenghui approved these changes Jul 13, 2021

View reviewed changes

codelipenghui added the release/2.7.3 label Jul 13, 2021

codelipenghui merged commit feb4ff1 into apache:master Jul 13, 2021

codelipenghui added release/2.7.4 and removed release/2.7.3 labels Jul 13, 2021

merlimat deleted the compaction-fixes branch July 13, 2021 17:11

codelipenghui added the cherry-picked/branch-2.8 Archived: 2.8 is end of life label Jul 14, 2021

Technoboy- added a commit to Technoboy-/pulsar that referenced this pull request Jul 20, 2021

Cherry-picking apache#11287 Fixed retention of keys in compaction in …

f1485cd

…branch-2.7.

Technoboy- added a commit to Technoboy-/pulsar that referenced this pull request Jul 21, 2021

Cherry-picking apache#11287 Fixed retention of keys in compaction in …

479b520

…branch-2.7.

Technoboy- added a commit to Technoboy-/pulsar that referenced this pull request Jul 23, 2021

Cherry-picking apache#11287 Fixed retention of keys in compaction in …

150463f

…branch-2.7 apache#11384

codelipenghui pushed a commit that referenced this pull request Jul 23, 2021

Cherry-picking #11287 Fixed retention of keys in compaction in branch…

d8b54d9

…-2.7 #11384 (#11384)

codelipenghui added the cherry-picked/branch-2.7 Archived: 2.7 is end of life label Jul 23, 2021

codelipenghui mentioned this pull request Aug 5, 2021

[WIP] Fix reading compacted data with the earliest start position by using Reader API #11159

Closed

merlimat mentioned this pull request Aug 15, 2021

Decouple original topic and compacted topic retention policies #6995

Open

codelipenghui mentioned this pull request Jan 5, 2022

Fix reader skipped remaining compacted data during the topic unloading. #13629

Merged

3 tasks

codelipenghui added the area/compaction label Feb 22, 2022

Technoboy- mentioned this pull request Jun 30, 2022

[Branch-2.7][Cherry-pick] Fix reader skipped remaining compacted data during the topic unloading #16300

Merged

3 tasks

coderzc mentioned this pull request Aug 28, 2023

[Bug] Topic unloading during Compaction leads to messages loss and messages redelivery loop #21074

Closed

2 tasks

coderzc mentioned this pull request Sep 18, 2023

[fix][broker] Avoid consumers receiving acknowledged messages from compacted topic after reconnection #21187

Merged

15 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fixed retention of keys in compaction #11287

Fixed retention of keys in compaction #11287

merlimat commented Jul 12, 2021

Anonymitaet commented Jul 13, 2021

codelipenghui left a comment

codelipenghui Jul 13, 2021

merlimat Jul 13, 2021

codelipenghui Jul 13, 2021

eolivelli left a comment


		assertTrue(expectedKeys.contains(msg.getKey()), "Unexpected key: " + msg.getKey());

		System.err.println("Received: " + msg.getKey());

Fixed retention of keys in compaction #11287

Fixed retention of keys in compaction #11287

Conversation

merlimat commented Jul 12, 2021

Motivation

Modifications

Anonymitaet commented Jul 13, 2021

codelipenghui left a comment

Choose a reason for hiding this comment

codelipenghui Jul 13, 2021

Choose a reason for hiding this comment

merlimat Jul 13, 2021

Choose a reason for hiding this comment

codelipenghui Jul 13, 2021

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment