Compacted topic will lose data when original topic message expired
- 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
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