IntenseFinance/nosqlbench
3
0 Fork 0
mirror of https://github.com/nosqlbench/nosqlbench.git synced 2026-10-12 17:32:01 -05:00
Code Issues Packages Projects Releases Wiki Activity

Optimize cache keys and fix races with caches in PulsarSpace

Browse Source
This commit is contained in:
Lari Hotari committed 2023-02-07 08:10:45 +02:00
1 parent 10da1699ed
commit 5491790751
3 files changed
+40 -89
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

adapter-pulsar/src/main/java/io/nosqlbench/adapter/pulsar/PulsarSpace.java
+18 -9
View File
@@ -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<String, Producer<?>> producers = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, Consumer<?>> consumers = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, Reader<?>> readers = new ConcurrentHashMap<>();
public record ProducerCacheKey(String producerName, String topicName) {
}
private final ConcurrentHashMap<ProducerCacheKey, Producer<?>> producers = new ConcurrentHashMap<>();
public record ConsumerCacheKey(String consumerName, String subscriptionName, List<String> topicNameList, String topicPattern) {
}
private final ConcurrentHashMap<ConsumerCacheKey, Consumer<?>> consumers = new ConcurrentHashMap<>();
public record ReaderCacheKey(String readerName, String topicName, String startMsgPosStr) {
}
private final ConcurrentHashMap<ReaderCacheKey, Reader<?>> 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<Producer<?>> 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<Consumer<?>> consumerSupplier) { return consumers.computeIfAbsent(key, __ -> consumerSupplier.get()); }
public Reader<?> getReader(ReaderCacheKey key, Supplier<Reader<?>> readerSupplier) { return readers.computeIfAbsent(key, __ -> readerSupplier.get()); }
/**
adapter-pulsar/src/main/java/io/nosqlbench/adapter/pulsar/dispensers/PulsarBaseOpDispenser.java
+22 -68
View File
@@ -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<PulsarOp, PulsarSpace> implements NBNamedElement {
@@ -239,10 +238,8 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser<PulsarOp, P
PulsarAdapterUtil.PRODUCER_CONF_STD_KEY.producerName.label,
cycleProducerName);
String producerCacheKey = PulsarAdapterUtil.buildCacheKey(producerName, topicName);
Producer<?> 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<PulsarOp, P
producerBuilder = producerBuilder.producerName(producerName);
}
producer = producerBuilder.create();
pulsarSpace.setProducer(producerCacheKey, producer);
Producer<?> 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<String> getEffectiveConsumerTopicNameList(String cycleTopicNameListStr) {
@@ -296,24 +289,6 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser<PulsarOp, P
return effectiveTopicNameList;
}
private Pattern getEffectiveConsumerTopicPattern(String cycleTopicPatternStr) {
String effectiveTopicPatternStr = getEffectiveConValue(
PulsarAdapterUtil.CONF_GATEGORY.Consumer.label,
PulsarAdapterUtil.CONSUMER_CONF_STD_KEY.topicsPattern.label,
cycleTopicPatternStr);
Pattern topicsPattern;
try {
if (!StringUtils.isBlank(effectiveTopicPatternStr))
topicsPattern = Pattern.compile(effectiveTopicPatternStr);
else
topicsPattern = null;
} catch (PatternSyntaxException pse) {
topicsPattern = null;
}
return topicsPattern;
}
private SubscriptionType getEffectiveSubscriptionType(String cycleSubscriptionType) {
String subscriptionTypeStr = getEffectiveConValue(
PulsarAdapterUtil.CONF_GATEGORY.Consumer.label,
@@ -344,11 +319,10 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser<PulsarOp, P
List<String> 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<PulsarOp, P
"creating multiple consumers of \"Exclusive\" subscription type under the same subscription name");
}
if ( (topicNameList.isEmpty() && (topicPattern == null)) ||
(!topicNameList.isEmpty() && (topicPattern != null)) ) {
if ( (topicNameList.isEmpty() && (topicPatternStr == null)) ||
(!topicNameList.isEmpty() && (topicPatternStr != null)) ) {
throw new PulsarAdapterInvalidParamException(
"Invalid combination of topic name(s) and topic patterns; only specify one parameter!");
}
boolean multiTopicConsumer = (topicNameList.size() > 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<PulsarOp, P
consumerConfToLoad.remove(PulsarAdapterUtil.CONSUMER_CONF_STD_KEY.negativeAckRedeliveryBackoff.label);
consumerConfToLoad.remove(PulsarAdapterUtil.CONSUMER_CONF_STD_KEY.ackTimeoutRedeliveryBackoff.label);
boolean multiTopicConsumer = (topicNameList.size() > 1 || (topicPatternStr != null));
if (!multiTopicConsumer) {
assert (topicNameList.size() == 1);
consumerBuilder = pulsarClient.newConsumer(pulsarSpace.getPulsarSchema());
@@ -429,6 +390,7 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser<PulsarOp, P
consumerBuilder.topics(topicNameList);
}
else {
Pattern topicPattern = Pattern.compile(topicPatternStr);
consumerBuilder.topicsPattern(topicPattern);
}
}
@@ -465,22 +427,22 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser<PulsarOp, P
consumerBuilder.keySharedPolicy(keySharedPolicy);
}
consumer = consumerBuilder.subscribe();
pulsarSpace.setConsumer(consumerCacheKey, consumer);
Consumer<?> 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<PulsarOp, P
throw new RuntimeException("Reader:: Invalid value for reader start message position!");
}
String readerCacheKey = PulsarAdapterUtil.buildCacheKey(topicName, readerName, startMsgPosStr);
Reader<?> reader = pulsarSpace.getReader(readerCacheKey);
if (reader == null) {
return pulsarSpace.getReader(new PulsarSpace.ReaderCacheKey(readerName, topicName, startMsgPosStr), () -> {
PulsarClient pulsarClient = pulsarSpace.getPulsarClient();;
Map<String, Object> readerConf = pulsarSpace.getPulsarNBClientConf().getReaderConfMapTgt();
@@ -558,17 +517,12 @@ public abstract class PulsarBaseOpDispenser extends BaseOpDispenser<PulsarOp, P
// startMsgId = MessageId.latest;
//}
reader = readerBuilder.startMessageId(startMsgId).create();
return readerBuilder.startMessageId(startMsgId).create();
} catch (PulsarClientException ple) {
ple.printStackTrace();
throw new RuntimeException("Unable to create a Pulsar reader!");
}
pulsarSpace.setReader(readerCacheKey, reader);
}
return reader;
});
}
//
//////////////////////////////////////
adapter-pulsar/src/main/java/io/nosqlbench/adapter/pulsar/util/PulsarAdapterUtil.java
-12
View File
@@ -482,18 +482,6 @@ public class PulsarAdapterUtil {
}
}
///////
// Generate effective key string
public static String buildCacheKey(String... keyParts) {
// Ignore blank keyPart
String joinedKeyStr =
Stream.of(keyParts)
.filter(s -> !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();