[improve][client] Add Builder Methods to Create Message-based TableView

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

@@ -40,6 +40,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiConsumer;
+import java.util.function.Function;
 import lombok.Cleanup;
 import lombok.CustomLog;
 import org.apache.commons.lang3.RandomUtils;
@@ -443,12 +444,12 @@ public void testAck(boolean partitionedTopic) throws Exception {
         if (partitionedTopic) {
             MultiTopicsReaderImpl<String> reader =
                     ((CompletableFuture<MultiTopicsReaderImpl<String>>) FieldUtils
-                            .readDeclaredField(tv1, "reader", true)).get();
+                            .readField(tv1, "reader", true)).get();
             consumerBase = spy(reader.getMultiTopicsConsumer());
             FieldUtils.writeDeclaredField(reader, "multiTopicsConsumer", consumerBase, true);
         } else {
             ReaderImpl<String> reader = ((CompletableFuture<ReaderImpl<String>>) FieldUtils
-                    .readDeclaredField(tv1, "reader", true)).get();
+                    .readField(tv1, "reader", true)).get();
             consumerBase = spy(reader.getConsumer());
             FieldUtils.writeDeclaredField(reader, "consumer", consumerBase, true);
         }
@@ -557,7 +558,7 @@ public void testTableViewTailMessageReadRetry() throws Exception {
 
         // inject failure on consumer.receiveAsync()
         var reader = ((CompletableFuture<Reader<byte[]>>)
-                FieldUtils.readDeclaredField(tv, "reader", true)).join();
+                FieldUtils.readField(tv, "reader", true)).join();
         var consumer = spy((ConsumerImpl<byte[]>)
                 FieldUtils.readDeclaredField(reader, "consumer", true));
 
@@ -625,7 +626,7 @@ public void testBuildTableViewWithMessagesAlwaysAvailable() throws Exception {
                 .createAsync()
                 .get();
         TableViewImpl<byte[]> mockTableView = spy(tableView);
-        Method readAllExistingMessagesMethod = TableViewImpl.class
+        Method readAllExistingMessagesMethod = AbstractTableViewImpl.class
                 .getDeclaredMethod("readAllExistingMessages", Reader.class);
         readAllExistingMessagesMethod.setAccessible(true);
         CompletableFuture<Reader<?>> future =
@@ -635,4 +636,86 @@ public void testBuildTableViewWithMessagesAlwaysAvailable() throws Exception {
         future.get(3, TimeUnit.SECONDS);
         assertTrue(index.get() <= 0);
     }
+
+    @Test
+    public void testCreateMapped() throws Exception {
+        String topic = "persistent://public/default/testCreateMapped";
+        admin.topics().createNonPartitionedTopic(topic);
+
+        @Cleanup
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+
+        @Cleanup
+        TableView<String> tableView = pulsarClient.newTableViewBuilder(Schema.STRING)
+                .topic(topic)
+                .createMapped(m -> {
+                    if (m.getValue().equals("delete-me")) {
+                        return null;
+                    }
+                    return m.getValue() + ":" + m.getProperty("myProp");
+                });
+
+        // Send a message to be mapped
+        String testKey = "key1";
+        producer.newMessage()
+                .key(testKey)
+                .value("value1")
+                .property("myProp", "myValue")
+                .send();
+
+        Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> tableView.size() == 1);
+        assertEquals(tableView.get(testKey), "value1:myValue");
+
+        // Send another message to update the value
+        producer.newMessage()
+                .key(testKey)
+                .value("value2")
+                .property("myProp", "myValue2")
+                .send();
+
+        Awaitility.await().atMost(5, TimeUnit.SECONDS)
+                .until(() -> "value2:myValue2".equals(tableView.get(testKey)));
+        assertEquals(tableView.size(), 1);
+
+        // Send a message that maps to null (tombstone)
+        producer.newMessage()
+                .key(testKey)
+                .value("delete-me")
+                .send();
+
+        Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> tableView.size() == 0);
+        Assert.assertNull(tableView.get(testKey), "Value should be null after tombstone message");
+    }
+
+    @Test
+    public void testCreateMappedWithIdentityMapper() throws Exception {
+        String topic = "persistent://public/default/testCreateMappedWithIdentityMapper";
+        admin.topics().createNonPartitionedTopic(topic);
+
+        @Cleanup
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+
+        String testKey = "key1";
+        String testValue = "value1";
+        producer.newMessage()
+                .key(testKey)
+                .value(testValue)
+                .property("myProp", "myValue")
+                .send();
+
+        @Cleanup
+        TableView<Message<String>> tableView = pulsarClient.newTableViewBuilder(Schema.STRING)
+                .topic(topic)
+                .createMapped(Function.identity());
+
+        Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> tableView.size() == 1);
+
+        Message<String> message = tableView.get(testKey);
+        Assert.assertNotNull(message, "Message should not be null for key: " + testKey);
+        assertEquals(message.getKey(), testKey);
+        assertEquals(message.getValue(), testValue);
+        assertEquals(message.getProperty("myProp"), "myValue");
+
+        Assert.assertNull(tableView.get("missingKey"), "Message should be null for missing key");
+    }
 }

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

@@ -336,7 +336,7 @@ public void testCompactionWithTableview() throws Exception {
                 .create();
 
         ((ServiceUnitStateDataConflictResolver)
-                FieldUtils.readDeclaredField(tv, "compactionStrategy", true))
+                FieldUtils.readField(tv, "compactionStrategy", true))
                 .checkBrokers(false);
         TestData testData = generateTestData();
         var topic = testData.topic;
@@ -679,7 +679,7 @@ public void testSlowReceiveTableviewAfterCompaction() throws Exception {
                 new StrategicTwoPhaseCompactor(conf, pulsarClient, bk, compactionScheduler);
 
         var reader = ((CompletableFuture<ReaderImpl<ServiceUnitStateData>>) FieldUtils
-                .readDeclaredField(tv, "reader", true)).get();
+                .readField(tv, "reader", true)).get();
         var consumer = spy(reader.getConsumer());
         FieldUtils.writeDeclaredField(reader, "consumer", consumer, true);
         String bundle = "bundle1";

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TableViewBuilder.java

@@ -21,6 +21,7 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 import org.apache.pulsar.common.classification.InterfaceAudience;
 import org.apache.pulsar.common.classification.InterfaceStability;
 
@@ -76,6 +77,48 @@ public interface TableViewBuilder<T> {
      */
     CompletableFuture<TableView<T>> createAsync();
 
+    /**
+     * Creates a {@link TableView} instance where the values are the result of applying a user-defined
+     * {@code mapper} function to each message.
+     *
+     * <p>This provides a flexible way to create a key-value view over a topic, allowing users to extract data
+     * from the message payload, properties, and other metadata into a custom object of type {@code V}.
+     *
+     * <p>To get a view of the full {@link Message} objects, {@code java.util.function.Function.identity()}
+     * can be used as the mapper. Message pooling is not used for mapped table views, so it is safe to keep
+     * a reference to the {@link Message} instance passed to the mapper.
+     *
+     * <p>If the {@code mapper} function returns {@code null}, it is treated as a tombstone message, and the
+     * corresponding key will be removed from the {@link TableView}.
+     *
+     * @param mapper a function that takes a {@link Message} and returns a custom object of type {@code V}
+     * @param <V> the type of the values in the {@link TableView}
+     * @return the {@link TableView} instance
+     * @throws PulsarClientException
+     *              if the tableView creation fails
+     */
+    <V> TableView<V> createMapped(Function<Message<T>, V> mapper) throws PulsarClientException;
+
+    /**
+     * Creates a {@link TableView} instance in asynchronous mode where the values are the result of applying
+     * a user-defined {@code mapper} function to each message.
+     *
+     * <p>This provides a flexible way to create a key-value view over a topic, allowing users to extract data
+     * from the message payload, properties, and other metadata into a custom object of type {@code V}.
+     *
+     * <p>To get a view of the full {@link Message} objects, {@code java.util.function.Function.identity()}
+     * can be used as the mapper. Message pooling is not used for mapped table views, so it is safe to keep
+     * a reference to the {@link Message} instance passed to the mapper.
+     *
+     * <p>If the {@code mapper} function returns {@code null}, it is treated as a tombstone message, and the
+     * corresponding key will be removed from the {@link TableView}.
+     *
+     * @param mapper a function that takes a {@link Message} and returns a custom object of type {@code V}
+     * @param <V> the type of the values in the {@link TableView}
+     * @return a future that can be used to access the {@link TableView} instance when it's ready
+     */
+    <V> CompletableFuture<TableView<V>> createMappedAsync(Function<Message<T>, V> mapper);
+
     /**
      * Set the topic name of the {@link TableView}.
      *