[fix][flaky-test]CompactedTopicTest.testReadCompleteMessagesDuringTopicUnloading

[poorbarcode]

Motivation

Task "Clear the ledger that has been consumed" will be triggered by mark delete event and retention policy.

(High light) Because there is no durable cursor(__compaction already acknowledged), the ledger will be removed by the asynchronous clean-ledger-task, even if the reader is still reading

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

Lines 754 to 778 in d4beae7

	admin.topics().triggerCompaction(topic);
	Awaitility.await().untilAsserted(() -> {
	PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic);
	Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1);
	Assert.assertEquals(stats.compactedLedger.entries, numMessages);
	Assert.assertEquals(admin.topics().getStats(topic)
	.getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0);
	Assert.assertEquals(stats.lastConfirmedEntry, stats.cursors.get(COMPACTION_SUBSCRIPTION).markDeletePosition);
	});
	// Unload the topic to make sure the original ledger been deleted.
	admin.topics().unload(topic);
	// Produce more messages to the original topic
	for (int i = 0; i < numMessages; ++i) {
	lastMessage = producer.newMessage().key(i + numMessages + "").value(String.format("msg [%d]", i + numMessages)).sendAsync();
	}
	producer.flush();
	lastMessage.join();
	// For now the topic has 1000 messages in the compacted ledger and 1000 messages in the original topic.
	@Cleanup
	Reader<String> reader = pulsarClient.newReader(Schema.STRING)
	.topic(topic)
	.startMessageIdInclusive()
	.startMessageId(MessageId.earliest)
	.readCompacted(true)
	.create();

---

The error execution flow is as follows:

producer:

• send 1000 messages ( ledger-1 )
• unload the topic.
• send 1000 messages (ledger-2)

reader:

• read 500 messages
• auto acknowledgment
• calculate the next read position:
  • getNextValidPosition(position) At this time the ledger-1 has been deleted, so getNextValidPosition(position) will return (originalLedger+1:0)
• read (ledger-2:0), then error occur

Modifications

• Create a consumer to generate a durable cursor, to avoid deleting the ledger before the reader receives.
• Make the reader creates before calling triggerCompaction

Documentation

• doc-required

• doc-not-needed

• doc

• doc-complete

[poorbarcode]

/pulsarbot rerun-failure-checks

[poorbarcode]

/pulsarbot rerun-failure-checks

[poorbarcode]

rebase master

[lhotari]

Thank you @poorbarcode