Ref: https://pulsar.apache.org/docs/next/concepts-clients/#tableview
TableView runs the registered listeners on each received message. TableView users could utilize TableView Listeners for event-sourcing applications.
Currently, TableView#foreachAndListen
and TableView#listen
interfaces are used to register listeners.
issue : #23158
As reported in the above issue, currently, the TableView listener registration methods, foreachAndListen
and listen
, are being called after TableView#start()
. This timing issue can result in some messages being missed or not handled if the listeners are registered too late.
Additionally, the current implementation of these listeners does not distinguish between existing messages (those present before the TableView started) and tail (future) messages (those received after the TableView started). Some business logic may require different handling for these two types of messages, but the current design does not support this differentiation.
- Support specifying TableView Listeners in TableView builder and constructor, so that TableView can register listeners as part of instance init and accordingly call them for all messages.
- Support specifying TableView Listeners for existing and tail messages separately.
- This design is based on Pulsar-Java-Client.
Add followings in TableViewBuilder.java
/**
* Set the message listeners.
* If {@link TableViewBuilder#existingMessageListeners} are not specified,
* these listeners are used for both
* existing and tailing(future) messages in the topic.
* @param listeners message listeners
* @return the {@link TableViewBuilder} builder instance
*/
TableViewBuilder<T> messageListeners(BiConsumer<String, T>... listeners);
/**
* Set the message listeners separately for existing messages in the topic.
* @param listeners message listeners
* @return the {@link TableViewBuilder} builder instance
*/
TableViewBuilder<T> existingMessageListeners(BiConsumer<String, T>... listeners);
Add followings in TableViewConfigurationData.java
private BiConsumer[] messageListeners;
private BiConsumer[] existingMessageListeners;
Add followings in TableViewImpl.java
private final List<BiConsumer<String, T>> existingMessageListeners;
Update followings in TableViewImpl.java
// before
private void handleMessage(Message<T> msg)
// after
private void handleMessage(Message<T> msg, boolean handleExistingMessage)
Compatible.
- Mailing List discussion thread:
- Mailing List voting thread: