From 54917907514cb9bb35267c11b3df20a1f08f223d Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Tue, 7 Feb 2023 08:06:42 +0200 Subject: [PATCH] Optimize cache keys and fix races with caches in PulsarSpace --- .../adapter/pulsar/PulsarSpace.java | 27 ++++-- .../dispensers/PulsarBaseOpDispenser.java | 90 +++++-------------- .../pulsar/util/PulsarAdapterUtil.java | 12 --- 3 files changed, 40 insertions(+), 89 deletions(-) diff --git a/adapter-pulsar/src/main/java/io/nosqlbench/adapter/pulsar/PulsarSpace.java b/adapter-pulsar/src/main/java/io/nosqlbench/adapter/pulsar/PulsarSpace.java index 4aba161f1..9cc5e44c6 100644 --- a/adapter-pulsar/src/main/java/io/nosqlbench/adapter/pulsar/PulsarSpace.java +++ b/adapter-pulsar/src/main/java/io/nosqlbench/adapter/pulsar/PulsarSpace.java @@ -32,8 +32,10 @@ import org.apache.pulsar.client.admin.PulsarAdminBuilder; import org.apache.pulsar.client.api.*; import org.apache.pulsar.common.schema.KeyValueEncodingType; +import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Supplier; public class PulsarSpace implements AutoCloseable { @@ -50,9 +52,18 @@ public class PulsarSpace implements AutoCloseable { private PulsarAdmin pulsarAdmin; private Schema pulsarSchema; - private final ConcurrentHashMap> producers = new ConcurrentHashMap<>(); - private final ConcurrentHashMap> consumers = new ConcurrentHashMap<>(); - private final ConcurrentHashMap> readers = new ConcurrentHashMap<>(); + public record ProducerCacheKey(String producerName, String topicName) { + } + + private final ConcurrentHashMap> producers = new ConcurrentHashMap<>(); + + public record ConsumerCacheKey(String consumerName, String subscriptionName, List topicNameList, String topicPattern) { + } + private final ConcurrentHashMap> consumers = new ConcurrentHashMap<>(); + + public record ReaderCacheKey(String readerName, String topicName, String startMsgPosStr) { + } + private final ConcurrentHashMap> readers = new ConcurrentHashMap<>(); public PulsarSpace(String spaceName, NBConfiguration cfg) { @@ -89,13 +100,11 @@ public class PulsarSpace implements AutoCloseable { public int getProducerSetCnt() { return producers.size(); } public int getConsumerSetCnt() { return consumers.size(); } public int getReaderSetCnt() { return readers.size(); } - public Producer getProducer(String name) { return producers.get(name); } - public void setProducer(String name, Producer producer) { producers.put(name, producer); } - public Consumer getConsumer(String name) { return consumers.get(name); } - public void setConsumer(String name, Consumer consumer) { consumers.put(name, consumer); } + public Producer getProducer(ProducerCacheKey key, Supplier> producerSupplier) { return producers.computeIfAbsent(key, __ -> producerSupplier.get()); } - public Reader getReader(String name) { return readers.get(name); } - public void setReader(String name, Reader reader) { readers.put(name, reader); } + public Consumer getConsumer(ConsumerCacheKey key, Supplier> consumerSupplier) { return consumers.computeIfAbsent(key, __ -> consumerSupplier.get()); } + + public Reader getReader(ReaderCacheKey key, Supplier> readerSupplier) { return readers.computeIfAbsent(key, __ -> readerSupplier.get()); } /** diff --git a/adapter-pulsar/src/main/java/io/nosqlbench/adapter/pulsar/dispensers/PulsarBaseOpDispenser.java b/adapter-pulsar/src/main/java/io/nosqlbench/adapter/pulsar/dispensers/PulsarBaseOpDispenser.java index 1de99a097..2e1b40230 100644 --- a/adapter-pulsar/src/main/java/io/nosqlbench/adapter/pulsar/dispensers/PulsarBaseOpDispenser.java +++ b/adapter-pulsar/src/main/java/io/nosqlbench/adapter/pulsar/dispensers/PulsarBaseOpDispenser.java @@ -37,7 +37,6 @@ import java.util.*; import java.util.function.LongFunction; import java.util.function.Predicate; import java.util.regex.Pattern; -import java.util.regex.PatternSyntaxException; import java.util.stream.Collectors; public abstract class PulsarBaseOpDispenser extends BaseOpDispenser implements NBNamedElement { @@ -239,10 +238,8 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser producer = pulsarSpace.getProducer(producerCacheKey); - - if (producer == null) { + PulsarSpace.ProducerCacheKey producerCacheKey = new PulsarSpace.ProducerCacheKey(producerName, topicName); + return pulsarSpace.getProducer(producerCacheKey, () -> { PulsarClient pulsarClient = pulsarSpace.getPulsarClient(); // Get other possible producer settings that are set at global level @@ -262,21 +259,17 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser producer = producerBuilder.create(); pulsarAdapterMetrics.registerProducerApiMetrics(producer, getPulsarAPIMetricsPrefix( PulsarAdapterUtil.PULSAR_API_TYPE.PRODUCER.label, producerName, topicName)); - } - catch (PulsarClientException ple) { + return producer; + } catch (PulsarClientException ple) { throw new PulsarAdapterUnexpectedException("Failed to create a Pulsar producer."); } - } - - return producer; + }); } private List getEffectiveConsumerTopicNameList(String cycleTopicNameListStr) { @@ -296,24 +289,6 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser topicNameList = getEffectiveConsumerTopicNameList(cycleTopicNameListStr); - String topicPatternStr = getEffectiveConValue( + String topicPatternStr = StringUtils.trimToNull(getEffectiveConValue( PulsarAdapterUtil.CONF_GATEGORY.Consumer.label, PulsarAdapterUtil.CONSUMER_CONF_STD_KEY.topicsPattern.label, - cycleTopicPatternStr); - Pattern topicPattern = getEffectiveConsumerTopicPattern(cycleTopicPatternStr); + cycleTopicPatternStr)); String subscriptionName = getEffectiveConValue( PulsarAdapterUtil.CONF_GATEGORY.Consumer.label, @@ -368,28 +342,14 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser 1 || (topicPattern != null)); - - String consumerTopicListString; - if (!topicNameList.isEmpty()) { - consumerTopicListString = String.join("|", topicNameList); - } else { - consumerTopicListString = topicPatternStr; - } - - String consumerCacheKey = PulsarAdapterUtil.buildCacheKey( - consumerName, - subscriptionName, - consumerTopicListString); - Consumer consumer = pulsarSpace.getConsumer(consumerCacheKey); - - if (consumer == null) { + return pulsarSpace.getConsumer( + new PulsarSpace.ConsumerCacheKey(consumerName, subscriptionName, topicNameList, topicPatternStr), () -> { PulsarClient pulsarClient = pulsarSpace.getPulsarClient(); // Get other possible consumer settings that are set at global level @@ -417,6 +377,7 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser 1 || (topicPatternStr != null)); if (!multiTopicConsumer) { assert (topicNameList.size() == 1); consumerBuilder = pulsarClient.newConsumer(pulsarSpace.getPulsarSchema()); @@ -429,6 +390,7 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser consumer = consumerBuilder.subscribe(); + String consumerTopicListString = (!topicNameList.isEmpty()) ? String.join("|", topicNameList) : topicPatternStr; pulsarAdapterMetrics.registerConsumerApiMetrics( consumer, getPulsarAPIMetricsPrefix( PulsarAdapterUtil.PULSAR_API_TYPE.CONSUMER.label, consumerName, consumerTopicListString)); + + return consumer; } catch (PulsarClientException ple) { throw new PulsarAdapterUnexpectedException("Failed to create a Pulsar consumer!"); } - } - - return consumer; + }); } private static Range[] parseRanges(String ranges) { @@ -528,10 +490,7 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser reader = pulsarSpace.getReader(readerCacheKey); - - if (reader == null) { + return pulsarSpace.getReader(new PulsarSpace.ReaderCacheKey(readerName, topicName, startMsgPosStr), () -> { PulsarClient pulsarClient = pulsarSpace.getPulsarClient();; Map readerConf = pulsarSpace.getPulsarNBClientConf().getReaderConfMapTgt(); @@ -558,17 +517,12 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser !StringUtils.isBlank(s)) - .collect(Collectors.joining(",")); - - return Base64.getEncoder().encodeToString(joinedKeyStr.getBytes()); - } - /////// // Convert JSON string to a key/value map private static final ObjectMapper JACKSON_OBJECT_MAPPER = new ObjectMapper();