[fix][client] Avert extensive time consumption during table view construction #21170

liangyepianzhou · 2023-09-12T13:23:31Z

Motivation

If a topic persistently experiences a substantial quantity of data inputs, the act of reading all the messages present in this topic to build a TableView can take an excessive amount of time.

Modification

In the process of constructing the TableView, initially, the last message ID of the current topic is procured. Consequently, once this last message ID has been reached, the creation ensues to its completion.

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (10MB)
Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository:

…truction ### Motivation If a topic persistently experiences a substantial quantity of data inputs, the act of reading all the messages present in this topic to build a TableView can take an excessive amount of time. ### Modification In the process of constructing the TableView, initially, the last message ID of the current topic is procured. Consequently, once this last message ID has been reached, the creation ensues to its completion.

Demogorgon314 · 2023-09-14T01:41:14Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java

@@ -235,20 +238,33 @@ private CompletableFuture<Reader<T>> readAllExistingMessages(Reader<T> reader) {
        AtomicLong messagesRead = new AtomicLong();

        CompletableFuture<Reader<T>> future = new CompletableFuture<>();
-        readAllExistingMessages(reader, future, startTime, messagesRead);
+        reader.getLastMessageIdsAsync().thenAccept(lastMessageIds -> {


I am wondering do we have a test to cover this?

I wrote a test, but I deleted it. This case is too extreme, and it might become wholly useless or unstable due to the testing machine and the configuration of the test data.
Do you have any ideas to test it?

Hi @Demogorgon314, I added a test by Mockito. Please take a look. Thanks.

michalcukierman · 2023-09-14T09:25:11Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java

@@ -235,20 +238,33 @@ private CompletableFuture<Reader<T>> readAllExistingMessages(Reader<T> reader) {
        AtomicLong messagesRead = new AtomicLong();

        CompletableFuture<Reader<T>> future = new CompletableFuture<>();
-        readAllExistingMessages(reader, future, startTime, messagesRead);
+        reader.getLastMessageIdsAsync().thenAccept(lastMessageIds -> {
+            Optional<TopicMessageId> optionalTopicMessageId = lastMessageIds.stream().max(Comparator.naturalOrder());


Some things to concern:

First case: working on partitioned topics:
Imagine:

topic-partiton-1 - containing 1 million messages

topic-partition-2 - containing 1 message

When a producer writes a new message to topic-partition-2 and it gets to TableView, the method will be stopped.

Second case: work with compaction

After the compaction, if the new client connects, the messages read first will have messageIds from the compacted ledger. Will compareTo work in that case?

Third case:

Comparison of MessageId when delivered in batches (this should probably be OK now)

I think that the algorithm should store every last message from each partition (first issue). Then the stop mechanism should be based on the exact message delivery.

Second case: work with compaction

After the compaction, if the new client connects, the messages read first will have messageIds from the compacted ledger. Will compareTo work in that case?

Messages read from a compacted topic by a normal client will be no different to messages read from a non-compacted topic, though there will be gaps in the message IDs. See https://github.com/apache/pulsar/wiki/PIP-14:-Topic-compaction

I think that the algorithm should store every last message from each partition (first issue). Then the stop mechanism should be based on the exact message delivery.

I totally agree. We only ensure that messages in the same partition are ordered, so the strategy of only judging the largest message ID is wrong. Thanks for your reminder.

Just to consider:
If the compacted ledger is created after the ledger that holds the backlog, it may have the higher ledger Id, right?

So you reed the latest message from the backlog

You start reading from the beginning (compacted ledger, first message)

The compacted ledger has higher ID, therefore the MessageId.equals won't work as expected

Is there a mechanism to prevent that?

Demogorgon314 · 2023-09-27T05:00:16Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java

+                                  if (maxMessageId != null && msg.getMessageId().compareTo(maxMessageId) >= 0) {
+                                      maxMessageIds.remove(msg.getTopicName());
+                                  }
+                                  if (maxMessageIds.size() == 0) {


Suggested change

if (maxMessageIds.size() == 0) {

if (maxMessageIds.isEmpty()) {

Demogorgon314 · 2023-09-27T05:07:02Z

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java

+        Method readAllExistingMessagesMethod = TableViewImpl.class
+                .getDeclaredMethod("readAllExistingMessages", Reader.class);
+        readAllExistingMessagesMethod.setAccessible(true);
+        CompletableFuture<Reader<?>> future =


Should we send more messages before invoking the readAllExistingMessages method to ensure we did not receive the new messages?

…truction (#21270) Reopen #21170 ### Motivation If a topic persistently experiences a substantial quantity of data inputs, the act of reading all the messages present in this topic to build a TableView can take an excessive amount of time. ### Modification In the process of constructing the TableView, initially, the last message ID of the current topic is procured. Consequently, once this last message ID has been reached, the creation ensues to its completion.

…truction (apache#21270) Reopen apache#21170 ### Motivation If a topic persistently experiences a substantial quantity of data inputs, the act of reading all the messages present in this topic to build a TableView can take an excessive amount of time. ### Modification In the process of constructing the TableView, initially, the last message ID of the current topic is procured. Consequently, once this last message ID has been reached, the creation ensues to its completion.

liangyepianzhou added 2 commits September 12, 2023 21:20

delete test

bbf1c7e

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Sep 12, 2023

liangyepianzhou mentioned this pull request Sep 12, 2023

[improve][client] PIP-302 Introduce refreshAsync API for TableView #21166

Closed

15 tasks

liangyepianzhou requested review from Technoboy-, nlu90, BewareMyPower, Demogorgon314 and poorbarcode September 13, 2023 06:56

liangyepianzhou self-assigned this Sep 13, 2023

Demogorgon314 reviewed Sep 14, 2023

View reviewed changes

michalcukierman reviewed Sep 14, 2023

View reviewed changes

address some comment and add test

e723102

Demogorgon314 reviewed Sep 27, 2023

View reviewed changes

liangyepianzhou closed this by deleting the head repository Sep 27, 2023

liangyepianzhou mentioned this pull request Sep 28, 2023

[fix][client] Avert extensive time consumption during table view construction #21270

Merged

15 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix][client] Avert extensive time consumption during table view construction #21170

[fix][client] Avert extensive time consumption during table view construction #21170

liangyepianzhou commented Sep 12, 2023

Demogorgon314 Sep 14, 2023

liangyepianzhou Sep 14, 2023

liangyepianzhou Sep 26, 2023

michalcukierman Sep 14, 2023

liangyepianzhou Sep 25, 2023

liangyepianzhou Sep 25, 2023

michalcukierman Sep 26, 2023

Demogorgon314 Sep 27, 2023

Demogorgon314 Sep 27, 2023

	if (maxMessageIds.size() == 0) {
	if (maxMessageIds.isEmpty()) {

[fix][client] Avert extensive time consumption during table view construction #21170

[fix][client] Avert extensive time consumption during table view construction #21170

Conversation

liangyepianzhou commented Sep 12, 2023

Motivation

Modification

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Matching PR in forked repository

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment