apache / apache/pulsar

Compacted topic will lose data when original topic message expired

Open
#11,211 1 comment 0 reactions 0 assignees View on GitHub
lifecycle/stale type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 22h
Merged PRs (30d)
142

Description

**Describe the bug**
When produce message to topic and trigger topic compaction, and then the original message expired by retention policy, the message which compacted into the compact ledger will also be lost.

If we set topic policy into `__change_events` topic and trigger compaction. And then the original policy in `__change_events` topic has been expired, the topic policy will lost.

The following test will show this case.

```Java
@Test(timeOut = 30000)
public void testCompactWithMessageTimeout() throws Exception {
final String topic = "persistent://my-property/use/my-ns/testCompactWithMessageTimeout-" + UUID.randomUUID();
admin.topics().createPartitionedTopic(topic, 1);

Producer producer = pulsarClient.newProducer(Schema.STRING).topic(topic).enableBatching(false).create();
// send 10 messages
for (int i = 0; i < 10; ++i) {
producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i)).send();
}

// trigger topic compaction
admin.topics().triggerCompaction(topic);
boolean succeed = retryStrategically((test) -> {
try {
return LongRunningProcessStatus.Status.SUCCESS.equals(admin.topics().compactionStatus(topic).status);
} catch (PulsarAdminException e) {
return false;
}
}, 10, 200);
Assert.assertTrue(succeed);

// unload topic to trigger ledger roll over
admin.topics().unload(topic);

// change ledger retention time and trim expired ledger to ensure the compacted message will be expired
PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topic).get();
ManagedLedgerConfig managedLedgerConfig = persistentTopic.getManagedLedger().getConfig();
managedLedgerConfig.setRetentionTime(1, TimeUnit.MILLISECONDS);
persistentTopic.getManagedLedger().trimConsumedLedgersInBackground(Futures.NULL_PROMISE);

// send another 10 messages
for (int i = 10; i < 20; ++i) {
producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i)).send();
}

// trigger topic compaction
admin.topics().triggerCompaction(topic);
succeed = retryStrategically((test) -> {
try {
return LongRunningProcessStatus.Status.SUCCESS.equals(admin.topics().compactionStatus(topic).status);
} catch (PulsarAdminException e) {
return false;
}
}, 10, 200);
Assert.assertTrue(succeed);

Reader reader = pulsarClient.newReader(Schema.STRING)
.topic(topic)
.subscriptionName("test")
.readCompacted(true)
.startMessageId(MessageId.earliest)
.create();

// check message in compacted topic, it should contains the whole messages
for (int i = 0; i < 20; ++i) {
Message msg = reader.readNext();
Assert.assertEquals(msg.getKey(), String.valueOf(i));
Assert.assertEquals(msg.getValue(), String.valueOf(i));
}
}
```

Contributor guide

Open the contributing guide

Research direction

Start with the testCompactWithMessageTimeout reproducer, then trace the compaction and ManagedLedger flow around PersistentTopic, ledger rollover, retention trimming, and readCompacted readers. Run the test to confirm the failure. Done means the second compaction preserves all 20 keyed messages after the original ledger has expired.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.