You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "liangyepianzhou (via GitHub)" <gi...@apache.org> on 2023/09/28 07:00:42 UTC
[GitHub] [pulsar] liangyepianzhou opened a new pull request, #21271: [improve][client] PIP-302 Introduce refreshAsync API for TableView
liangyepianzhou opened a new pull request, #21271:
URL: https://github.com/apache/pulsar/pull/21271
## Motivation
**Prerequisite:** Since messages are constantly being written into the Topic and there is no read-write lock guarantee, we cannot assure the retrieval of the most up-to-date value.
**Implementation Goal:** Record a checkpoint before reading and ensure the retrieval of the latest value of the key up to this checkpoint.
**Use Case:** When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
## Modification
Introduce a new API `refreshAsync`.
### 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:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#issuecomment-1751941749
>
> I think the main concern is that `get` might be blocking while Pulsar uses the `future.xxxAsync` everywhere and it's dangerous to call a blocking method in the callback of `xxxAsync`. For the sake of this reason, I'm +1 to add a method to read messages to latest at a certain time point.
>
> However, I think the `refreshAsync` design is not good. Considering the following case:
>
> ```java
> view.refreshAsync().thenAccept(__ -> {
> // Now, the TableView is the latest snapshot of the compacted topic
> if (view.containsKey(key)) { // T1
> process(view.get(key)); // T2
> }
> });
> ```
>
> Assume at `T1`, the internal data is `{ key => value1 }`. Since the `TableView` keeps fetching the latest message at the background, if a new message `(key, value2)` is written after `T1` and before `T2`, at `T2`, `view.get(key)` will be `value2`.
>
> The root cause is that `TableView` is not immutable. `TableViewImpl#readTailMessages` fetches the messages at the background. So I suggest returning an immutable map like:
>
> ```java
> /**
> * Get the latest snapshot for a set of keys at the current time point.
> *
> * @param keys
> * @return a future of the map that represents the snapshot of the table view. The keys must be a subset of the
> * `keys` parameter and the value is guaranteed to be the latest value before the future is completed.
> */
> CompletableFuture<Map<String, T>> getLatestSnapshotAsync(Set<String> keys);
>
> /**
> * Get the latest snapshot at the current time point.
> * @return a future of map that represents the snapshot of the table view.
> */
> CompletableFuture<Map<String, T>> getLatestSnapshotAsync();
> ```
The current use case is that when reads and writes for a key do not occur at the same time, we can refresh the TableView next time to get the latest value of this key.
As for the `getLatestSnapshotAsync` method, I don't quite understand its use case. Could you give an example to illustrate under what circumstances we would use this guarantee?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349646909
##########
pip/pip-302.md:
##########
@@ -0,0 +1,104 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
+}
+```
+
+# Public-Facing Changes
+
+## Public API
+
+The following changes will be made to the public API:
+
+### `refreshAsync()`
+
+This new API method triggers the reading of all existing messages, updates the TableView, and waits for the read operation to complete.
+```java
+/**
+ * Triggers the reading of all existing messages from the topics, updates the TableView and waits for the read operation to complete.
+ * This method fetches the last message of the topics at the point of invocation and updates the TableView with all messages up to and including this last message.
+ * After the update is complete, users can use the TableView to obtain the latest value for any key.
+ * Note that the 'latest' value refers to the value at the point of calling refresh, not necessarily the current latest if more messages have been produced in the meantime.
+ *
+ * @return a CompletableFuture that completes when all existing messages up to the point of invocation have been read, and the TableView has been updated.
+ *
+ * Example usage:
+ * table.refreshAsync().whenComplete((__, ex) -> {
+ * if (ex == null) {
+ * table.get(key);
+ * }
+ * });
Review Comment:
Why doesn't `whenComplete` work? And don't we need to handle exceptions?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349643927
##########
pip/pip-302.md:
##########
@@ -0,0 +1,104 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
+}
+```
+
+# Public-Facing Changes
+
+## Public API
+
+The following changes will be made to the public API:
+
+### `refreshAsync()`
+
+This new API method triggers the reading of all existing messages, updates the TableView, and waits for the read operation to complete.
+```java
+/**
+ * Triggers the reading of all existing messages from the topics, updates the TableView and waits for the read operation to complete.
+ * This method fetches the last message of the topics at the point of invocation and updates the TableView with all messages up to and including this last message.
+ * After the update is complete, users can use the TableView to obtain the latest value for any key.
+ * Note that the 'latest' value refers to the value at the point of calling refresh, not necessarily the current latest if more messages have been produced in the meantime.
+ *
+ * @return a CompletableFuture that completes when all existing messages up to the point of invocation have been read, and the TableView has been updated.
+ *
+ * Example usage:
+ * table.refreshAsync().whenComplete((__, ex) -> {
+ * if (ex == null) {
+ * table.get(key);
+ * }
+ * });
Review Comment:
Why do we need to use `table.refreshAsync().thenApply(__ -> table.get(key));` instead of
```
* table.refreshAsync().whenComplete((__, ex) -> {
* if (ex == null) {
* table.get(key);
* }
* });
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349811240
##########
pip/pip-302.md:
##########
@@ -0,0 +1,100 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
+}
+```
+
+# Public-Facing Changes
+
+## Public API
+
+The following changes will be made to the public API:
+
+### `refreshAsync()`
+
+This new API method triggers the reading of all existing messages, updates the TableView, and waits for the read operation to complete.
+```java
+/**
+ * Triggers the reading of all existing messages from the topics, updates the TableView and waits for the read operation to complete.
+ * This method fetches the last message of the topics at the point of invocation and updates the TableView with all messages up to and including this last message.
+ * After the update is complete, users can use the TableView to obtain the latest value for any key.
+ * Note that the 'latest' value refers to the value at the point of calling refresh, not necessarily the current latest if more messages have been produced in the meantime.
+ *
+ * @return a CompletableFuture that completes when all existing messages up to the point of invocation have been read, and the TableView has been updated.
+ *
+ * Example usage:
+ * table.refreshAsync().thenApply(__ -> table.get(key));
+ */
Review Comment:
```suggestion
/**
*
* Refresh the table view with the latest data in the topic, ensuring that all subsequent reads are based on the refreshed data.
*
* Example usage:
*
* table.refreshAsync().thenApply(__ -> table.get(key));
*
* This function retrieves the last written message in the topic and refreshes the table view accordingly.
* Once the refresh is complete, all subsequent reads will be performed on the refreshed data or a combination of the refreshed
* data and newly published data. The table view remains synchronized with any newly published data after the refresh.
*
* |x:0|->|y:0|->|z:0|->|x:1|->|z:1|->|x:2|->|y:1|->|y:2|
*
* If a read occurs after the refresh (at the last published message |y:2|), it ensures that outdated data like x=1 is not obtained.
* However, it does not guarantee that the values will always be x=2, y=2, z=1, as the table view may receive updates with newly
* published data.
*
* |x:0|->|y:0|->|z:0|->|x:1|->|z:1|->|x:2|->|y:1|->|y:2| ---> |y:3|
*
* Both y=2 or y=3 are possible. Therefore, different readers may receive different values, but all values will be equal to or newer
* than the data refreshed from the last call to the refresh method.
*/
```
I would like to provide another description of the method to make sure we can explain what we are guaranteed and what is not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349644820
##########
pip/pip-302.md:
##########
@@ -0,0 +1,104 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
+}
+```
+
+# Public-Facing Changes
+
+## Public API
+
+The following changes will be made to the public API:
+
+### `refreshAsync()`
+
+This new API method triggers the reading of all existing messages, updates the TableView, and waits for the read operation to complete.
+```java
+/**
+ * Triggers the reading of all existing messages from the topics, updates the TableView and waits for the read operation to complete.
+ * This method fetches the last message of the topics at the point of invocation and updates the TableView with all messages up to and including this last message.
+ * After the update is complete, users can use the TableView to obtain the latest value for any key.
+ * Note that the 'latest' value refers to the value at the point of calling refresh, not necessarily the current latest if more messages have been produced in the meantime.
+ *
+ * @return a CompletableFuture that completes when all existing messages up to the point of invocation have been read, and the TableView has been updated.
+ *
+ * Example usage:
+ * table.refreshAsync().whenComplete((__, ex) -> {
+ * if (ex == null) {
+ * table.get(key);
+ * }
+ * });
Review Comment:
Take the example above, if you immediately call `refreshAsync` after sending 10000 messages rather than in the callback of `lastSendFuture`, the behavior is undefined.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou merged PR #21271:
URL: https://github.com/apache/pulsar/pull/21271
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349641879
##########
pip/pip-302.md:
##########
@@ -0,0 +1,104 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
Review Comment:
There is a `readTailMessages` loop at the background.`readAllExistingMessages` could also call `readNextAsync`. Is there any possible race?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349642124
##########
pip/pip-302.md:
##########
@@ -0,0 +1,104 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
Review Comment:
This is a matter of specific implementation. We can optimize this during the implementation process.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#issuecomment-1751943721
> As for the getLatestSnapshotAsync method, I don't quite understand its use case.
It provides a consistency view. For example, let's assume there are two keys that represents x and y that satisfy `y = x * 2` (just an example). The change might be:
- x: 0, 1, 3, 7, ...
- y: 0, 2, 6, 14, ...
With `refreshAsync`, you might call:
```java
view.refreshAsync().thenAccept(__ -> { // now, x = 3 and y = 6
int x = view.get("x"); // x = 3
int y = view.get("y"); // y = 14, actually, at the moment, view.get("x") is 7
// If users want to process x and y here, they might find the difference is unexpected large
```
With `getLatestSnapshotAsync()` method, you can write:
```java
view.getLatestSnapshotAsync().thenAccept(snapshot -> {
int x = snapshot.get("x"); // x = 3, view.get("x") = 3, view.get("y") = 6
int y = snapshot.get("y"); // y = 6, even if now view.get("y") is 14
});
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349644106
##########
pip/pip-302.md:
##########
@@ -0,0 +1,104 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
+}
+```
+
+# Public-Facing Changes
+
+## Public API
+
+The following changes will be made to the public API:
+
+### `refreshAsync()`
+
+This new API method triggers the reading of all existing messages, updates the TableView, and waits for the read operation to complete.
+```java
+/**
+ * Triggers the reading of all existing messages from the topics, updates the TableView and waits for the read operation to complete.
+ * This method fetches the last message of the topics at the point of invocation and updates the TableView with all messages up to and including this last message.
+ * After the update is complete, users can use the TableView to obtain the latest value for any key.
+ * Note that the 'latest' value refers to the value at the point of calling refresh, not necessarily the current latest if more messages have been produced in the meantime.
+ *
+ * @return a CompletableFuture that completes when all existing messages up to the point of invocation have been read, and the TableView has been updated.
+ *
+ * Example usage:
+ * table.refreshAsync().whenComplete((__, ex) -> {
+ * if (ex == null) {
+ * table.get(key);
+ * }
+ * });
Review Comment:
A better example might be like:
```java
CompletableFuture<MessageId> lastSendFuture = null;
for (int i = 0; i < 10000; i++) {
lastSendFuture = producer.newMessage().key("key-" + (i % 5)).value(("msg-" + i).getBytes()).sendAsync();
}
lastSendFuture.thenCompose(__ -> tableView.refreshAsync()).thenAccept(__ -> {
final byte[] value = tableView.get("key-1");
// ...
});
```
i.e. all producers are controlled by yourself. You know there is no more message written to the topic, then it will be safe and meaningful to call `refreshAsync()` and other methods like `get()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349643198
##########
pip/pip-302.md:
##########
@@ -0,0 +1,104 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
+}
+```
+
+# Public-Facing Changes
+
+## Public API
+
+The following changes will be made to the public API:
+
+### `refreshAsync()`
+
+This new API method triggers the reading of all existing messages, updates the TableView, and waits for the read operation to complete.
+```java
+/**
+ * Triggers the reading of all existing messages from the topics, updates the TableView and waits for the read operation to complete.
+ * This method fetches the last message of the topics at the point of invocation and updates the TableView with all messages up to and including this last message.
+ * After the update is complete, users can use the TableView to obtain the latest value for any key.
+ * Note that the 'latest' value refers to the value at the point of calling refresh, not necessarily the current latest if more messages have been produced in the meantime.
+ *
+ * @return a CompletableFuture that completes when all existing messages up to the point of invocation have been read, and the TableView has been updated.
+ *
+ * Example usage:
+ * table.refreshAsync().whenComplete((__, ex) -> {
+ * if (ex == null) {
+ * table.get(key);
+ * }
+ * });
Review Comment:
```suggestion
* table.refreshAsync().thenApply(__ -> table.get(key));
```
Make the code example more meaningful. Maybe you can add the comment that use the code above instead of the trivial `table.get(key)`.
Actually, I think it's not much different from a `getAsync` method because now you return a `CompletableFuture<T>` rather than `T`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349649112
##########
pip/pip-302.md:
##########
@@ -0,0 +1,104 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
+}
+```
+
+# Public-Facing Changes
+
+## Public API
+
+The following changes will be made to the public API:
+
+### `refreshAsync()`
+
+This new API method triggers the reading of all existing messages, updates the TableView, and waits for the read operation to complete.
+```java
+/**
+ * Triggers the reading of all existing messages from the topics, updates the TableView and waits for the read operation to complete.
+ * This method fetches the last message of the topics at the point of invocation and updates the TableView with all messages up to and including this last message.
+ * After the update is complete, users can use the TableView to obtain the latest value for any key.
+ * Note that the 'latest' value refers to the value at the point of calling refresh, not necessarily the current latest if more messages have been produced in the meantime.
+ *
+ * @return a CompletableFuture that completes when all existing messages up to the point of invocation have been read, and the TableView has been updated.
+ *
+ * Example usage:
+ * table.refreshAsync().whenComplete((__, ex) -> {
+ * if (ex == null) {
+ * table.get(key);
+ * }
+ * });
Review Comment:
We can add `exceptionally` or `get`/`join` to handle the exception. `whenComplete` works but the use is not direct. Without the `refreshAsync`, the use of `TableView` is very simple and direct.
For example, assuming the application handles a request from client like:
```java
T handleQuery(Request req) {
return tableView.get(req.key());
}
```
All the calls are synchronous so they are easy to use.
However, with a `whenComplete` callback, you can only process the key in the callback.
```java
T handleQuery(Request req) {
var future = new CompletableFuture<T>();
tableView.refreshAsync().whenComplete((__, e) -> {
if (e != null) {
future.complete(tableView.get(req.key()));
}
});
return future.join(); // change it to synchronous
}
```
The code above is complicated, and it might be hard to users that get used to synchronous programming. For example, they might also write code like:
```java
T handleQuery(Request req) throws InterruptedException {
var value = new AtomicReference<T>();
tableView.refreshAsync().whenComplete((__, e) -> {
if (e != null) {
value.set(tableView.get(req.key()));
}
});
while (value.get() == null) {
Thread.sleep(1);
}
return value.get();
}
```
In the previous use case, `thenApply` is more clear and simple.
```java
T handleQuery(Request req) {
return tableView.thenApply(__ -> tableView.get(req.key()).join();
}
```
If any exception is thrown, `join()` (or `get()`) will throw it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#issuecomment-1751899833
I see @merlimat's suggestion [here](https://github.com/apache/pulsar/pull/21166#issuecomment-1719694532) and you respond: https://github.com/apache/pulsar/pull/21166#issuecomment-1720430117
> Even if we refresh the tableView before each data read, we can't guarantee that the data retrieved is the most up-to-date.
With this proposal, we still cannot guarantee the "latest" value is retrieved. Just like you mentioned, this proposal only guarantees the value is latest at the checkpoint when `refreshAsync` is called.
> This configuration might confuse users.
I disagree. Assume we added two options: `STRONG_CONSISTENCY_MODEL` and `EVENTUAL_CONSISTENCY_MODEL`), then we'll have:
- `STRONG_CONSISTENCY_MODEL`: any method will be blocked until the latest value is retrieved.
- `EVENTUAL_CONSISTENCY_MODEL`: all methods are non-blocking but the value retrieved might not be latest at the time point.
This pattern is similar to the [POSIX `read` API](https://man7.org/linux/man-pages/man2/read.2.html), which is widely used by system development. It's blocking by default, i.e. it will be blocked until some data is available. But if the file descriptor is marked as `O_NONBLOCK`, it will become non-blocking.
> We would need to add many asynchronous interfaces, such as T getAsync(String key); boolean containsKeyAsync(String key); Collection valuesAsync(); etc.
I think the main concern is that `get` might be blocking while Pulsar uses the `future.xxxAsync` everywhere and it's dangerous to call a blocking method in the callback of `xxxAsync`. For the sake of this reason, I'm +1 to add a method to read messages to latest at a certain time point.
However, I think the `refreshAsync` design is not good. Considering the following case:
```java
view.refreshAsync().thenAccept(__ -> {
// Now, the TableView is the latest snapshot of the compacted topic
if (view.containsKey(key)) { // T1
process(view.get(key)); // T2
}
});
```
Assume at `T1`, the internal data is `{ key => value1 }`. Since the `TableView` keeps fetching the latest message at the background, if a new message `(key, value2)` is written after `T1` and before `T2`, at `T2`, `view.get(key)` will be `value2`.
The root cause is that `TableView` is not immutable. `TableViewImpl#readTailMessages` fetches the messages at the background. So I suggest returning an immutable map like:
```java
/**
* Get the latest snapshot for a set of keys at the current time point.
*
* @param keys
* @return a future of the map that represents the snapshot of the table view. The keys must be a subset of the
* `keys` parameter and the value is guaranteed to be the latest value before the future is completed.
*/
CompletableFuture<Map<String, T>> getLatestSnapshotAsync(Set<String> keys);
/**
* Get the latest snapshot at the current time point.
* @return a future of map that represents the snapshot of the table view.
*/
CompletableFuture<Map<String, T>> getLatestSnapshotAsync();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#issuecomment-1751946996
>If you can control the write side, why not just use a Reader?
Because there is a system topic shared by all topics in this namespace.
If we use Reader, then all topics may need to read the system topic from beginning to end when it loads.
If we use TableView, the system topic only needs to be read one time from beginning to end for each broker.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#issuecomment-1751949561
>If `refreshAsync` method is just to handle this case, you must note the limitation of `refreshAsync` that it cannot guarantee latest value is read unless there is no messages sent to the topic before the future is complete. i.e. it's meaningful only when the caller takes control of all producers to the topic.
Yes, It will be clarified in the API comments.
It should be noted that this is actually the greatest assurance we can provide right now. Because there might be continuous messages being sent, and the rate of sending messages might be faster than the rate of receiving messages, we actually can't guarantee to get the latest message. All we can do is help you refresh once and get the latest value at the current point in time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#issuecomment-1751940433
> > This configuration might confuse users.
>
> I disagree. Assume we added two options: `STRONG_CONSISTENCY_MODEL` and `EVENTUAL_CONSISTENCY_MODEL`), then we'll have:
> - `STRONG_CONSISTENCY_MODEL`: any method will be blocked until the latest value is retrieved.
> - `EVENTUAL_CONSISTENCY_MODEL`: all methods are non-blocking but the value retrieved might not be latest at the time point.
>
Here's a point to note: because there might be continuous message writing, we can't guarantee getting the latest value. If we provide users with the `STRONG_CONSISTENCY_MODEL` option, they might assume that they get the latest value every time, but this is incorrect.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349642943
##########
pip/pip-302.md:
##########
@@ -0,0 +1,104 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
Review Comment:
We should not touch the very concrete implementation in the proposal. But we should illustrate "how" to implement it. You should remove the `readAllExistingMessages` call here because nobody knows the method's semantics unless reading the source code. Instead, you should write the implementation like "waiting until there is more messages via `hasMessageAvailable`" or "the returned future will be completed when there is no more message".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#issuecomment-1751947747
> If we use Reader, then all topics may need to read the system topic from beginning to end when it loads.
Yeah I just also thought of this point so I deleted my comment before.
> The current use case is that when reads and writes for a key do not occur at the same time,
If this `refreshAsync` method is just to handle the case, you must note the limitation that it cannot guarantee latest value is read unless there is no messages sent to the topic before the future is complete. i.e. it's meaningful only when the caller takes control of all producers to the topic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349651345
##########
pip/pip-302.md:
##########
@@ -0,0 +1,104 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
+}
+```
+
+# Public-Facing Changes
+
+## Public API
+
+The following changes will be made to the public API:
+
+### `refreshAsync()`
+
+This new API method triggers the reading of all existing messages, updates the TableView, and waits for the read operation to complete.
+```java
+/**
+ * Triggers the reading of all existing messages from the topics, updates the TableView and waits for the read operation to complete.
+ * This method fetches the last message of the topics at the point of invocation and updates the TableView with all messages up to and including this last message.
+ * After the update is complete, users can use the TableView to obtain the latest value for any key.
+ * Note that the 'latest' value refers to the value at the point of calling refresh, not necessarily the current latest if more messages have been produced in the meantime.
+ *
+ * @return a CompletableFuture that completes when all existing messages up to the point of invocation have been read, and the TableView has been updated.
+ *
+ * Example usage:
+ * table.refreshAsync().whenComplete((__, ex) -> {
+ * if (ex == null) {
+ * table.get(key);
+ * }
+ * });
Review Comment:
Okay
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#discussion_r1349642545
##########
pip/pip-302.md:
##########
@@ -0,0 +1,104 @@
+# Background Knowledge
+
+The TableView interface provides a convenient way to access data in a compacted topic by offering a continuously updated key-value map view. It operates on messages with keys and ignores those without keys.
+
+By utilizing the TableView, Pulsar clients can retrieve all message updates from a specific topic and create a map that contains the latest values for each key. This map can be used to establish a local cache of data. Additionally, clients can register consumers with the TableView and specify a listener to scan the map and receive notifications whenever new messages are received. This functionality enables event-driven applications and message monitoring.
+
+For more detailed information about the TableView, please refer to the [Pulsar documentation](https://pulsar.apache.org/docs/next/concepts-clients/#tableview).
+
+# Motivation
+
+In the context of utilizing the TableView component, there are instances where we aspire to consistently retrieve the most up-to-date value associated with a given key.
+When read and write operations for a certain key do not occur simultaneously, we can refresh the TableView before reading the key to obtain the latest value for this key.
+To accomplish this, we can employ an API `refreshAsync` that allows us to refresh the TableView before accessing the value corresponding to the desired key.
+# Goals
+
+## In Scope
+
+The proposed changes aim to add a new API that triggers the reading of all existing messages that before the trigger time,
+updates the TableView, and waits for the read operation to complete. This will be achieved by introducing the `refreshAsync()` method.
+
+## Out of Scope
+
+No out-of-scope items have been identified at this time.
+
+# High-Level Design
+
+The high-level design involves the addition of a new API method called `refreshAsync()`.
+This method will be responsible for initiating the process of reading all existing messages that before the trigger time,
+updating the TableView, and returning a future to wait for the read operation to complete.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+The `refreshAsync()` method will be implemented as follows:
+
+```java
+@Override
+public CompletableFuture<Void> refreshAsync() {
+ return reader.thenCompose(this::readAllExistingMessages);
Review Comment:
I think we only need to call `hasMessageAvailableAsync` until the future returns false.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][pip] PIP-302 Introduce refreshAsync API for TableView [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#issuecomment-1751944806
> when reads and writes for a key do not occur at the same time, we can refresh the TableView next time to get the latest value of this key.
If you can control the write side, why not just use a `Reader`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #21271: [improve][client] PIP-302 Introduce refreshAsync API for TableView
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21271:
URL: https://github.com/apache/pulsar/pull/21271#issuecomment-1738590647
@liangyepianzhou Please add the following content to your PR description and select a checkbox:
```
- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org