diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java index 3408606e1ad77..02c17855b5a71 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TableViewTest.java +++ b/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 reader = ((CompletableFuture>) FieldUtils - .readDeclaredField(tv1, "reader", true)).get(); + .readField(tv1, "reader", true)).get(); consumerBase = spy(reader.getMultiTopicsConsumer()); FieldUtils.writeDeclaredField(reader, "multiTopicsConsumer", consumerBase, true); } else { ReaderImpl reader = ((CompletableFuture>) 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>) - FieldUtils.readDeclaredField(tv, "reader", true)).join(); + FieldUtils.readField(tv, "reader", true)).join(); var consumer = spy((ConsumerImpl) FieldUtils.readDeclaredField(reader, "consumer", true)); @@ -625,7 +626,7 @@ public void testBuildTableViewWithMessagesAlwaysAvailable() throws Exception { .createAsync() .get(); TableViewImpl mockTableView = spy(tableView); - Method readAllExistingMessagesMethod = TableViewImpl.class + Method readAllExistingMessagesMethod = AbstractTableViewImpl.class .getDeclaredMethod("readAllExistingMessages", Reader.class); readAllExistingMessagesMethod.setAccessible(true); CompletableFuture> 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 producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create(); + + @Cleanup + TableView 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 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> tableView = pulsarClient.newTableViewBuilder(Schema.STRING) + .topic(topic) + .createMapped(Function.identity()); + + Awaitility.await().atMost(5, TimeUnit.SECONDS).until(() -> tableView.size() == 1); + + Message 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"); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/ServiceUnitStateCompactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/ServiceUnitStateCompactionTest.java index 136ba751397fd..ca2f5b85818b0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/ServiceUnitStateCompactionTest.java +++ b/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>) 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"; diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TableViewBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TableViewBuilder.java index 76b8ff4fbdac3..f88a76c043d52 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TableViewBuilder.java +++ b/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 { */ CompletableFuture> createAsync(); + /** + * Creates a {@link TableView} instance where the values are the result of applying a user-defined + * {@code mapper} function to each message. + * + *

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}. + * + *

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. + * + *

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 the type of the values in the {@link TableView} + * @return the {@link TableView} instance + * @throws PulsarClientException + * if the tableView creation fails + */ + TableView createMapped(Function, 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. + * + *

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}. + * + *

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. + * + *

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 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 + */ + CompletableFuture> createMappedAsync(Function, V> mapper); + /** * Set the topic name of the {@link TableView}. * diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractTableViewImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractTableViewImpl.java new file mode 100644 index 0000000000000..f83c282e0c233 --- /dev/null +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractTableViewImpl.java @@ -0,0 +1,460 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.client.impl; + +import static org.apache.pulsar.common.topics.TopicCompactionStrategy.TABLE_VIEW_TAG; +import io.github.merlimat.slog.Logger; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.BiConsumer; +import org.apache.pulsar.client.api.CryptoKeyReader; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.MessageIdAdv; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Reader; +import org.apache.pulsar.client.api.ReaderBuilder; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.TableView; +import org.apache.pulsar.client.api.TopicMessageId; +import org.apache.pulsar.common.naming.TopicDomain; +import org.apache.pulsar.common.topics.TopicCompactionStrategy; + +/** + * Base class for {@link TableView} implementations. It reads messages of the schema type {@code T} + * from the topic and maintains a map of the latest value of type {@code V} for each key. + * Subclasses define how a message is converted into the value stored in the view. + * + * @param the message schema type + * @param the type of the values stored in the view + */ +abstract class AbstractTableViewImpl implements TableView { + + private static final Logger LOG = Logger.get(AbstractTableViewImpl.class); + private final Logger log; + private final TableViewConfigurationData conf; + + private final ConcurrentMap data; + private final Map immutableData; + + private final CompletableFuture> reader; + + private final List> listeners; + private final ReentrantLock listenersMutex; + private final boolean isPersistentTopic; + private final boolean poolMessages; + private TopicCompactionStrategy compactionStrategy; + + /** + * Store the refresh tasks. When read to the position recording in the right map, + * then remove the position in the right map. If the right map is empty, complete the future in the left. + * There should be no timeout exception here, because the caller can only retry for TimeoutException. + * It will only be completed exceptionally when no more messages can be read. + */ + private final ConcurrentHashMap, Map> pendingRefreshRequests; + + /** + * This map stored the read position of each partition. It is used for the following case: + *

+ * 1. Get last message ID. + * 2. Receive message p1-1:1, p2-1:1, p2-1:2, p3-1:1 + * 3. Receive response of step1 {|p1-1:1|p2-2:2|p3-3:6|} + * 4. No more messages are written to this topic. + * As a result, the refresh operation will never be completed. + *

+ */ + private final ConcurrentHashMap lastReadPositions; + + /** + * @param poolMessages whether the reader should use pooled messages. When enabled, the handled messages + * are released after they have been processed, so subclasses must not let the + * message instance escape from {@link #getValue(Message)}. + */ + AbstractTableViewImpl(PulsarClientImpl client, Schema schema, TableViewConfigurationData conf, + boolean poolMessages) { + this.conf = conf; + this.log = LOG.with().attr("topic", conf.getTopicName()).build(); + this.poolMessages = poolMessages; + this.isPersistentTopic = conf.getTopicName().startsWith(TopicDomain.persistent.toString()); + this.data = new ConcurrentHashMap<>(); + this.immutableData = Collections.unmodifiableMap(data); + this.listeners = new ArrayList<>(); + this.listenersMutex = new ReentrantLock(); + this.compactionStrategy = + TopicCompactionStrategy.load(TABLE_VIEW_TAG, conf.getTopicCompactionStrategyClassName()); + this.pendingRefreshRequests = new ConcurrentHashMap<>(); + this.lastReadPositions = new ConcurrentHashMap<>(); + ReaderBuilder readerBuilder = client.newReader(schema) + .topic(conf.getTopicName()) + .startMessageId(MessageId.earliest) + .autoUpdatePartitions(true) + .autoUpdatePartitionsInterval((int) conf.getAutoUpdatePartitionsSeconds(), TimeUnit.SECONDS) + .poolMessages(poolMessages) + .subscriptionName(conf.getSubscriptionName()); + if (isPersistentTopic) { + readerBuilder.readCompacted(true); + } + + CryptoKeyReader cryptoKeyReader = conf.getCryptoKeyReader(); + if (cryptoKeyReader != null) { + readerBuilder.cryptoKeyReader(cryptoKeyReader); + } + + readerBuilder.cryptoFailureAction(conf.getCryptoFailureAction()); + + this.reader = readerBuilder.createAsync(); + } + + CompletableFuture> start() { + return reader.thenCompose((reader) -> { + if (!isPersistentTopic) { + readTailMessages(reader); + return CompletableFuture.completedFuture(null); + } + return this.readAllExistingMessages(reader) + .thenRun(() -> readTailMessages(reader)); + }).thenApply(__ -> this); + } + + @Override + public int size() { + return data.size(); + } + + @Override + public boolean isEmpty() { + return data.isEmpty(); + } + + @Override + public boolean containsKey(String key) { + return data.containsKey(key); + } + + @Override + public V get(String key) { + return data.get(key); + } + + @Override + public Set> entrySet() { + return immutableData.entrySet(); + } + + @Override + public Set keySet() { + return immutableData.keySet(); + } + + @Override + public Collection values() { + return immutableData.values(); + } + + @Override + public void forEach(BiConsumer action) { + data.forEach(action); + } + + @Override + public void listen(BiConsumer action) { + try { + listenersMutex.lock(); + listeners.add(action); + } finally { + listenersMutex.unlock(); + } + } + + @Override + public void forEachAndListen(BiConsumer action) { + // Ensure we iterate over all the existing entry _and_ start the listening from the exact next message + try { + listenersMutex.lock(); + + // Execute the action over existing entries + forEach(action); + + listeners.add(action); + } finally { + listenersMutex.unlock(); + } + } + + @Override + public CompletableFuture closeAsync() { + return reader.thenCompose(Reader::closeAsync); + } + + @Override + public void close() throws PulsarClientException { + try { + closeAsync().get(); + } catch (Exception e) { + throw PulsarClientException.unwrap(e); + } + } + + private void handleMessage(Message msg) { + lastReadPositions.put(msg.getTopicName(), msg.getMessageId()); + try { + if (msg.hasKey()) { + String key = msg.getKey(); + V cur = getValueIfPresent(msg); + log.debug().attr("key", key) + .attr("value", cur) + .log("Applying message"); + + boolean update = true; + if (compactionStrategy != null) { + V prev = data.get(key); + update = !compactionStrategy.shouldKeepLeft(prev, cur); + if (!update) { + log.info().attr("key", key) + .attr("value", cur) + .attr("prev", prev) + .log("Skipped the message"); + compactionStrategy.handleSkippedMessage(key, cur); + } + } + + if (update) { + try { + listenersMutex.lock(); + if (null == cur) { + data.remove(key); + } else { + data.put(key, cur); + } + + for (BiConsumer listener : listeners) { + try { + listener.accept(key, cur); + } catch (Throwable t) { + log.error().exception(t).log("Table view listener raised an exception"); + } + } + } finally { + listenersMutex.unlock(); + } + } + } + checkAllFreshTask(msg); + } finally { + if (poolMessages) { + msg.release(); + } + } + } + + private V getValueIfPresent(Message msg) { + return msg.size() > 0 ? getValue(msg) : null; + } + + /** + * Converts the message into the value stored in the view. Only called for messages with a non-empty + * payload; messages with an empty payload are tombstones and remove the key from the view. + * A {@code null} return value is also handled as a tombstone. + * + * @param msg the message to convert + * @return the value to store in the view, or {@code null} to remove the key + */ + protected abstract V getValue(Message msg); + + @Override + public CompletableFuture refreshAsync() { + CompletableFuture completableFuture = new CompletableFuture<>(); + reader.thenCompose(reader -> getLastMessageIdOfNonEmptyTopics(reader).thenAccept(lastMessageIds -> { + if (lastMessageIds.isEmpty()) { + completableFuture.complete(null); + return; + } + // After get the response of lastMessageIds, put the future and result into `refreshMap` + // and then filter out partitions that has been read to the lastMessageID. + pendingRefreshRequests.put(completableFuture, lastMessageIds); + filterReceivedMessages(lastMessageIds); + // If there is no new messages, the refresh operation could be completed right now. + if (lastMessageIds.isEmpty()) { + pendingRefreshRequests.remove(completableFuture); + completableFuture.complete(null); + } + })).exceptionally(throwable -> { + completableFuture.completeExceptionally(throwable); + pendingRefreshRequests.remove(completableFuture); + return null; + }); + return completableFuture; + } + + @Override + public void refresh() throws PulsarClientException { + try { + refreshAsync().get(); + } catch (Exception e) { + throw PulsarClientException.unwrap(e); + } + } + + private CompletableFuture readAllExistingMessages(Reader reader) { + long startTime = System.nanoTime(); + AtomicLong messagesRead = new AtomicLong(); + + CompletableFuture future = new CompletableFuture<>(); + getLastMessageIdOfNonEmptyTopics(reader).thenAccept(lastMessageIds -> { + if (lastMessageIds.isEmpty()) { + future.complete(null); + return; + } + readAllExistingMessages(reader, future, startTime, messagesRead, lastMessageIds); + }).exceptionally(ex -> { + future.completeExceptionally(ex); + return null; + }); + return future; + } + + private CompletableFuture> getLastMessageIdOfNonEmptyTopics(Reader reader) { + return reader.getLastMessageIdsAsync().thenApply(lastMessageIds -> { + Map lastMessageIdMap = new ConcurrentHashMap<>(); + lastMessageIds.forEach(topicMessageId -> { + if (((MessageIdAdv) topicMessageId).getEntryId() >= 0) { + lastMessageIdMap.put(topicMessageId.getOwnerTopic(), topicMessageId); + } // else: a negative entry id represents an empty topic so that we don't have to read messages from it + }); + return lastMessageIdMap; + }); + } + + private void filterReceivedMessages(Map lastMessageIds) { + // The `lastMessageIds` and `readPositions` is concurrency-safe data types. + lastMessageIds.forEach((partition, lastMessageId) -> { + MessageId messageId = lastReadPositions.get(partition); + if (messageId != null && lastMessageId.compareTo(messageId) <= 0) { + lastMessageIds.remove(partition); + } + }); + } + + private boolean checkFreshTask(Map maxMessageIds, CompletableFuture future, + MessageId messageId, String topicName) { + // The message received from multi-consumer/multi-reader is processed to TopicMessageImpl. + TopicMessageId maxMessageId = maxMessageIds.get(topicName); + // We need remove the partition from the maxMessageIds map + // once the partition has been read completely. + if (maxMessageId != null && messageId.compareTo(maxMessageId) >= 0) { + maxMessageIds.remove(topicName); + } + if (maxMessageIds.isEmpty()) { + future.complete(null); + return true; + } else { + return false; + } + } + + private void checkAllFreshTask(Message msg) { + pendingRefreshRequests.forEach((future, maxMessageIds) -> { + String topicName = msg.getTopicName(); + MessageId messageId = msg.getMessageId(); + if (checkFreshTask(maxMessageIds, future, messageId, topicName)) { + pendingRefreshRequests.remove(future); + } + }); + } + + private void readAllExistingMessages(Reader reader, CompletableFuture future, long startTime, + AtomicLong messagesRead, Map maxMessageIds) { + reader.hasMessageAvailableAsync() + .thenAccept(hasMessage -> { + if (hasMessage) { + reader.readNextAsync() + .thenAccept(msg -> { + messagesRead.incrementAndGet(); + String topicName = msg.getTopicName(); + MessageId messageId = msg.getMessageId(); + handleMessage(msg); + if (!checkFreshTask(maxMessageIds, future, messageId, topicName)) { + readAllExistingMessages(reader, future, startTime, + messagesRead, maxMessageIds); + } + }).exceptionally(ex -> { + if (ex.getCause() instanceof PulsarClientException.AlreadyClosedException) { + log.info().attr("reader", reader.getTopic()) + .log("Reader was closed while reading existing messages."); + } else { + log.warn().attr("reader", reader.getTopic()) + .exception(ex) + .log("Reader was interrupted while reading existing messages."); + } + future.completeExceptionally(ex); + return null; + }); + } else { + // Reached the end + long endTime = System.nanoTime(); + long durationMillis = TimeUnit.NANOSECONDS.toMillis(endTime - startTime); + log.info().attr("topic", reader.getTopic()) + .attr("replayed", messagesRead) + .attr("durationSeconds", durationMillis / 1000.0) + .log("Started table view for topic - Replayed messages"); + future.complete(null); + } + }); + } + + private void readTailMessages(Reader reader) { + reader.readNextAsync() + .thenAccept(msg -> { + handleMessage(msg); + readTailMessages(reader); + }).exceptionally(ex -> { + if (ex.getCause() instanceof PulsarClientException.AlreadyClosedException) { + log.info().attr("reader", reader.getTopic()) + .log("Reader was closed while reading tail messages."); + // Fail all refresh request when no more messages can be read. + pendingRefreshRequests.keySet().forEach(future -> { + pendingRefreshRequests.remove(future); + future.completeExceptionally(ex); + }); + } else { + // Retrying on the other exceptions such as NotConnectedException + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + log.warn().attr("reader", reader.getTopic()) + .exception(ex) + .log("Reader was interrupted while reading tail messages. " + "Retrying.."); + readTailMessages(reader); + } + return null; + }); + } +} diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageMapperTableViewImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageMapperTableViewImpl.java new file mode 100644 index 0000000000000..6a8f8ef91babd --- /dev/null +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageMapperTableViewImpl.java @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.client.impl; + +import java.util.function.Function; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.Schema; + +/** + * {@link org.apache.pulsar.client.api.TableView} implementation that applies a user-provided mapper + * function to each message to produce the value stored in the view. + * + * @param the message schema type + * @param the value type returned by the mapper function + */ +public class MessageMapperTableViewImpl extends AbstractTableViewImpl { + + private final Function, V> mapper; + + MessageMapperTableViewImpl(PulsarClientImpl client, Schema schema, TableViewConfigurationData conf, + Function, V> mapper) { + // The message instance is passed to the user-provided mapper function, which may keep a reference + // to it (e.g. when Function.identity() is used as the mapper). Pooled messages must not be used + // since there is no way to know when the message could be released. + super(client, schema, conf, false); + this.mapper = mapper; + } + + @Override + protected V getValue(Message msg) { + return mapper.apply(msg); + } +} diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewBuilderImpl.java index e0a47a70b1c8b..9b44d07314243 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewBuilderImpl.java @@ -22,15 +22,18 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import java.util.function.Function; import lombok.NonNull; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.client.api.ConsumerCryptoFailureAction; import org.apache.pulsar.client.api.CryptoKeyReader; +import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.TableView; import org.apache.pulsar.client.api.TableViewBuilder; import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils; +import org.apache.pulsar.common.util.FutureUtil; public class TableViewBuilderImpl implements TableViewBuilder { @@ -65,6 +68,24 @@ public CompletableFuture> createAsync() { return new TableViewImpl<>(client, schema, conf).start(); } + @Override + public TableView createMapped(Function, V> mapper) throws PulsarClientException { + checkArgument(mapper != null, "mapper cannot be null"); + try { + return createMappedAsync(mapper).get(); + } catch (Exception e) { + throw PulsarClientException.unwrap(e); + } + } + + @Override + public CompletableFuture> createMappedAsync(Function, V> mapper) { + if (mapper == null) { + return FutureUtil.failedFuture(new IllegalArgumentException("mapper cannot be null")); + } + return new MessageMapperTableViewImpl<>(client, schema, conf, mapper).start(); + } + @Override public TableViewBuilder topic(String topic) { checkArgument(StringUtils.isNotBlank(topic), "topic cannot be blank"); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java index 8ed4e61a7cb43..257b4c4cea67d 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TableViewImpl.java @@ -18,411 +18,24 @@ */ package org.apache.pulsar.client.impl; -import static org.apache.pulsar.common.topics.TopicCompactionStrategy.TABLE_VIEW_TAG; -import io.github.merlimat.slog.Logger; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.locks.ReentrantLock; -import java.util.function.BiConsumer; -import org.apache.pulsar.client.api.CryptoKeyReader; import org.apache.pulsar.client.api.Message; -import org.apache.pulsar.client.api.MessageId; -import org.apache.pulsar.client.api.MessageIdAdv; -import org.apache.pulsar.client.api.PulsarClientException; -import org.apache.pulsar.client.api.Reader; -import org.apache.pulsar.client.api.ReaderBuilder; import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.api.TableView; -import org.apache.pulsar.client.api.TopicMessageId; -import org.apache.pulsar.common.naming.TopicDomain; -import org.apache.pulsar.common.topics.TopicCompactionStrategy; -public class TableViewImpl implements TableView { - - private static final Logger LOG = Logger.get(TableViewImpl.class); - private final Logger log; - private final TableViewConfigurationData conf; - - private final ConcurrentMap data; - private final Map immutableData; - - private final CompletableFuture> reader; - - private final List> listeners; - private final ReentrantLock listenersMutex; - private final boolean isPersistentTopic; - private TopicCompactionStrategy compactionStrategy; - - /** - * Store the refresh tasks. When read to the position recording in the right map, - * then remove the position in the right map. If the right map is empty, complete the future in the left. - * There should be no timeout exception here, because the caller can only retry for TimeoutException. - * It will only be completed exceptionally when no more messages can be read. - */ - private final ConcurrentHashMap, Map> pendingRefreshRequests; - - /** - * This map stored the read position of each partition. It is used for the following case: - *

- * 1. Get last message ID. - * 2. Receive message p1-1:1, p2-1:1, p2-1:2, p3-1:1 - * 3. Receive response of step1 {|p1-1:1|p2-2:2|p3-3:6|} - * 4. No more messages are written to this topic. - * As a result, the refresh operation will never be completed. - *

- */ - private final ConcurrentHashMap lastReadPositions; +/** + * {@link org.apache.pulsar.client.api.TableView} implementation that stores the deserialized message + * payload as the value. + * + * @param the message schema type + */ +public class TableViewImpl extends AbstractTableViewImpl { TableViewImpl(PulsarClientImpl client, Schema schema, TableViewConfigurationData conf) { - this.conf = conf; - this.log = LOG.with().attr("topic", conf.getTopicName()).build(); - this.isPersistentTopic = conf.getTopicName().startsWith(TopicDomain.persistent.toString()); - this.data = new ConcurrentHashMap<>(); - this.immutableData = Collections.unmodifiableMap(data); - this.listeners = new ArrayList<>(); - this.listenersMutex = new ReentrantLock(); - this.compactionStrategy = - TopicCompactionStrategy.load(TABLE_VIEW_TAG, conf.getTopicCompactionStrategyClassName()); - this.pendingRefreshRequests = new ConcurrentHashMap<>(); - this.lastReadPositions = new ConcurrentHashMap<>(); - ReaderBuilder readerBuilder = client.newReader(schema) - .topic(conf.getTopicName()) - .startMessageId(MessageId.earliest) - .autoUpdatePartitions(true) - .autoUpdatePartitionsInterval((int) conf.getAutoUpdatePartitionsSeconds(), TimeUnit.SECONDS) - .poolMessages(true) - .subscriptionName(conf.getSubscriptionName()); - if (isPersistentTopic) { - readerBuilder.readCompacted(true); - } - - CryptoKeyReader cryptoKeyReader = conf.getCryptoKeyReader(); - if (cryptoKeyReader != null) { - readerBuilder.cryptoKeyReader(cryptoKeyReader); - } - - readerBuilder.cryptoFailureAction(conf.getCryptoFailureAction()); - - this.reader = readerBuilder.createAsync(); - } - - CompletableFuture> start() { - return reader.thenCompose((reader) -> { - if (!isPersistentTopic) { - readTailMessages(reader); - return CompletableFuture.completedFuture(null); - } - return this.readAllExistingMessages(reader) - .thenRun(() -> readTailMessages(reader)); - }).thenApply(__ -> this); - } - - @Override - public int size() { - return data.size(); - } - - @Override - public boolean isEmpty() { - return data.isEmpty(); - } - - @Override - public boolean containsKey(String key) { - return data.containsKey(key); - } - - @Override - public T get(String key) { - return data.get(key); - } - - @Override - public Set> entrySet() { - return immutableData.entrySet(); - } - - @Override - public Set keySet() { - return immutableData.keySet(); - } - - @Override - public Collection values() { - return immutableData.values(); - } - - @Override - public void forEach(BiConsumer action) { - data.forEach(action); + // the message is fully consumed while extracting the value, so pooled messages can be used and released + super(client, schema, conf, true); } @Override - public void listen(BiConsumer action) { - try { - listenersMutex.lock(); - listeners.add(action); - } finally { - listenersMutex.unlock(); - } - } - - @Override - public void forEachAndListen(BiConsumer action) { - // Ensure we iterate over all the existing entry _and_ start the listening from the exact next message - try { - listenersMutex.lock(); - - // Execute the action over existing entries - forEach(action); - - listeners.add(action); - } finally { - listenersMutex.unlock(); - } - } - - @Override - public CompletableFuture closeAsync() { - return reader.thenCompose(Reader::closeAsync); - } - - @Override - public void close() throws PulsarClientException { - try { - closeAsync().get(); - } catch (Exception e) { - throw PulsarClientException.unwrap(e); - } - } - - private void handleMessage(Message msg) { - lastReadPositions.put(msg.getTopicName(), msg.getMessageId()); - try { - if (msg.hasKey()) { - String key = msg.getKey(); - T cur = msg.size() > 0 ? msg.getValue() : null; - log.debug().attr("key", key) - .attr("value", cur) - .log("Applying message"); - - boolean update = true; - if (compactionStrategy != null) { - T prev = data.get(key); - update = !compactionStrategy.shouldKeepLeft(prev, cur); - if (!update) { - log.info().attr("key", key) - .attr("value", cur) - .attr("prev", prev) - .log("Skipped the message"); - compactionStrategy.handleSkippedMessage(key, cur); - } - } - - if (update) { - try { - listenersMutex.lock(); - if (null == cur) { - data.remove(key); - } else { - data.put(key, cur); - } - - for (BiConsumer listener : listeners) { - try { - listener.accept(key, cur); - } catch (Throwable t) { - log.error().exception(t).log("Table view listener raised an exception"); - } - } - } finally { - listenersMutex.unlock(); - } - } - } - checkAllFreshTask(msg); - } finally { - msg.release(); - } - } - - @Override - public CompletableFuture refreshAsync() { - CompletableFuture completableFuture = new CompletableFuture<>(); - reader.thenCompose(reader -> getLastMessageIdOfNonEmptyTopics(reader).thenAccept(lastMessageIds -> { - if (lastMessageIds.isEmpty()) { - completableFuture.complete(null); - return; - } - // After get the response of lastMessageIds, put the future and result into `refreshMap` - // and then filter out partitions that has been read to the lastMessageID. - pendingRefreshRequests.put(completableFuture, lastMessageIds); - filterReceivedMessages(lastMessageIds); - // If there is no new messages, the refresh operation could be completed right now. - if (lastMessageIds.isEmpty()) { - pendingRefreshRequests.remove(completableFuture); - completableFuture.complete(null); - } - })).exceptionally(throwable -> { - completableFuture.completeExceptionally(throwable); - pendingRefreshRequests.remove(completableFuture); - return null; - }); - return completableFuture; - } - - @Override - public void refresh() throws PulsarClientException { - try { - refreshAsync().get(); - } catch (Exception e) { - throw PulsarClientException.unwrap(e); - } - } - - private CompletableFuture readAllExistingMessages(Reader reader) { - long startTime = System.nanoTime(); - AtomicLong messagesRead = new AtomicLong(); - - CompletableFuture future = new CompletableFuture<>(); - getLastMessageIdOfNonEmptyTopics(reader).thenAccept(lastMessageIds -> { - if (lastMessageIds.isEmpty()) { - future.complete(null); - return; - } - readAllExistingMessages(reader, future, startTime, messagesRead, lastMessageIds); - }).exceptionally(ex -> { - future.completeExceptionally(ex); - return null; - }); - return future; - } - - private CompletableFuture> getLastMessageIdOfNonEmptyTopics(Reader reader) { - return reader.getLastMessageIdsAsync().thenApply(lastMessageIds -> { - Map lastMessageIdMap = new ConcurrentHashMap<>(); - lastMessageIds.forEach(topicMessageId -> { - if (((MessageIdAdv) topicMessageId).getEntryId() >= 0) { - lastMessageIdMap.put(topicMessageId.getOwnerTopic(), topicMessageId); - } // else: a negative entry id represents an empty topic so that we don't have to read messages from it - }); - return lastMessageIdMap; - }); - } - - private void filterReceivedMessages(Map lastMessageIds) { - // The `lastMessageIds` and `readPositions` is concurrency-safe data types. - lastMessageIds.forEach((partition, lastMessageId) -> { - MessageId messageId = lastReadPositions.get(partition); - if (messageId != null && lastMessageId.compareTo(messageId) <= 0) { - lastMessageIds.remove(partition); - } - }); - } - - private boolean checkFreshTask(Map maxMessageIds, CompletableFuture future, - MessageId messageId, String topicName) { - // The message received from multi-consumer/multi-reader is processed to TopicMessageImpl. - TopicMessageId maxMessageId = maxMessageIds.get(topicName); - // We need remove the partition from the maxMessageIds map - // once the partition has been read completely. - if (maxMessageId != null && messageId.compareTo(maxMessageId) >= 0) { - maxMessageIds.remove(topicName); - } - if (maxMessageIds.isEmpty()) { - future.complete(null); - return true; - } else { - return false; - } - } - - private void checkAllFreshTask(Message msg) { - pendingRefreshRequests.forEach((future, maxMessageIds) -> { - String topicName = msg.getTopicName(); - MessageId messageId = msg.getMessageId(); - if (checkFreshTask(maxMessageIds, future, messageId, topicName)) { - pendingRefreshRequests.remove(future); - } - }); - } - - private void readAllExistingMessages(Reader reader, CompletableFuture future, long startTime, - AtomicLong messagesRead, Map maxMessageIds) { - reader.hasMessageAvailableAsync() - .thenAccept(hasMessage -> { - if (hasMessage) { - reader.readNextAsync() - .thenAccept(msg -> { - messagesRead.incrementAndGet(); - String topicName = msg.getTopicName(); - MessageId messageId = msg.getMessageId(); - handleMessage(msg); - if (!checkFreshTask(maxMessageIds, future, messageId, topicName)) { - readAllExistingMessages(reader, future, startTime, - messagesRead, maxMessageIds); - } - }).exceptionally(ex -> { - if (ex.getCause() instanceof PulsarClientException.AlreadyClosedException) { - log.info().attr("reader", reader.getTopic()) - .log("Reader was closed while reading existing messages."); - } else { - log.warn().attr("reader", reader.getTopic()) - .exception(ex) - .log("Reader was interrupted while reading existing messages."); - } - future.completeExceptionally(ex); - return null; - }); - } else { - // Reached the end - long endTime = System.nanoTime(); - long durationMillis = TimeUnit.NANOSECONDS.toMillis(endTime - startTime); - log.info().attr("topic", reader.getTopic()) - .attr("replayed", messagesRead) - .attr("durationSeconds", durationMillis / 1000.0) - .log("Started table view for topic - Replayed messages"); - future.complete(null); - } - }); - } - - private void readTailMessages(Reader reader) { - reader.readNextAsync() - .thenAccept(msg -> { - handleMessage(msg); - readTailMessages(reader); - }).exceptionally(ex -> { - if (ex.getCause() instanceof PulsarClientException.AlreadyClosedException) { - log.info().attr("reader", reader.getTopic()) - .log("Reader was closed while reading tail messages."); - // Fail all refresh request when no more messages can be read. - pendingRefreshRequests.keySet().forEach(future -> { - pendingRefreshRequests.remove(future); - future.completeExceptionally(ex); - }); - } else { - // Retrying on the other exceptions such as NotConnectedException - try { - Thread.sleep(50); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - log.warn().attr("reader", reader.getTopic()) - .exception(ex) - .log("Reader was interrupted while reading tail messages. " + "Retrying.."); - readTailMessages(reader); - } - return null; - }); + protected T getValue(Message msg) { + return msg.getValue(); } } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewBuilderImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewBuilderImplTest.java index b4bdda3a84077..0331dc38c7b33 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewBuilderImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewBuilderImplTest.java @@ -22,12 +22,14 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; import java.util.HashMap; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.api.ConsumerCryptoFailureAction; import org.apache.pulsar.client.api.CryptoKeyReader; +import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.client.api.Schema; @@ -147,4 +149,25 @@ public void testTableViewImplWhenDefaultCryptoKeyReaderIsNullMap() throws Pulsar public void testTableViewImplWhenDefaultCryptoKeyReaderIsEmptyMap() throws PulsarClientException { tableViewBuilderImpl.topic(TOPIC_NAME).defaultCryptoKeyReader(new HashMap()).create(); } + + @Test + public void testCreateMapped() throws PulsarClientException { + TableView tableView = tableViewBuilderImpl.topic(TOPIC_NAME) + .createMapped(Message::getKey); + + assertNotNull(tableView); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testCreateMappedWhenMapperIsNull() throws PulsarClientException { + tableViewBuilderImpl.topic(TOPIC_NAME).createMapped(null); + } + + @Test + public void testCreateMappedAsyncWhenMapperIsNull() { + CompletableFuture> future = + tableViewBuilderImpl.topic(TOPIC_NAME).createMappedAsync(null); + + assertTrue(future.isCompletedExceptionally()); + } }