diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarConsumerSpace.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarConsumerSpace.java
new file mode 100644
index 000000000..edf99267e
--- /dev/null
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarConsumerSpace.java
@@ -0,0 +1,180 @@
+package io.nosqlbench.driver.pulsar;
+
+import io.nosqlbench.driver.pulsar.util.PulsarActivityUtil;
+import io.nosqlbench.driver.pulsar.util.PulsarNBClientConf;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+public class PulsarConsumerSpace extends PulsarSpace {
+
+    private final ConcurrentHashMap<String, Consumer<?>> consumers = new ConcurrentHashMap<>();
+
+    public PulsarConsumerSpace(String name, PulsarNBClientConf pulsarClientConf) { super(name, pulsarClientConf); }
+
+    private String getEffectiveTopicNamesStr(String cycleTopicNames) {
+        if ((cycleTopicNames != null) && (!cycleTopicNames.isEmpty())) {
+            return cycleTopicNames;
+        }
+
+        String globalTopicNames = pulsarNBClientConf.getConsumerTopicNames();
+        if ((globalTopicNames != null) && (!globalTopicNames.isEmpty())) {
+            return globalTopicNames;
+        }
+
+        return "";
+    }
+    private List<String> getEffectiveTopicNames(String cycleTopicNames) {
+        String effectiveTopicNamesStr = getEffectiveTopicNamesStr(cycleTopicNames);
+
+        String[] names = effectiveTopicNamesStr.split("[;,]");
+        ArrayList<String> effectiveTopicNameList = new ArrayList<>();
+
+        for (String name : names) {
+            if ( !name.isEmpty() )
+                effectiveTopicNameList.add(name.trim());
+        }
+
+
+        return effectiveTopicNameList;
+    }
+
+    private String getEffectiveTopicPatternStr(String cycleTopicsPattern) {
+        if ((cycleTopicsPattern != null) && (!cycleTopicsPattern.isEmpty())) {
+            return cycleTopicsPattern;
+        }
+
+        String globalTopicsPattern = pulsarNBClientConf.getConsumerTopicPattern();
+        if ((globalTopicsPattern != null) && (!globalTopicsPattern.isEmpty())) {
+            return globalTopicsPattern;
+        }
+
+        return "";
+    }
+    private Pattern getEffectiveTopicPattern(String cycleTopicsPattern) {
+        String effecitveTopicsPatternStr = getEffectiveTopicPatternStr(cycleTopicsPattern);
+        Pattern topicsPattern;
+        try {
+            topicsPattern = Pattern.compile(effecitveTopicsPatternStr);
+        }
+        catch (PatternSyntaxException pse) {
+            topicsPattern = null;
+        }
+        return topicsPattern;
+    }
+
+    private String getEffectiveSubscriptionName(String cycleSubscriptionName) {
+        if ((cycleSubscriptionName != null) && (!cycleSubscriptionName.isEmpty())) {
+            return cycleSubscriptionName;
+        }
+
+        String globalSubscriptionName = pulsarNBClientConf.getConsumerSubscriptionName();
+        if ((globalSubscriptionName != null) && (!globalSubscriptionName.isEmpty())) {
+            return globalSubscriptionName;
+        }
+
+        return "default-subs";
+    }
+
+    private String getEffectiveSubscriptionTypeStr(String cycleSubscriptionType) {
+        if ((cycleSubscriptionType != null) && (!cycleSubscriptionType.isEmpty())) {
+            return cycleSubscriptionType;
+        }
+
+        String globalSubscriptionType = pulsarNBClientConf.getConsumerSubscriptionType();
+        if ((globalSubscriptionType != null) && (!globalSubscriptionType.isEmpty())) {
+            return globalSubscriptionType;
+        }
+
+        return "";
+    }
+    private SubscriptionType getEffectiveSubscriptionType(String cycleSubscriptionType) {
+        String effectiveSubscriptionStr = getEffectiveSubscriptionTypeStr(cycleSubscriptionType);
+        SubscriptionType subscriptionType;
+
+        try {
+            subscriptionType = SubscriptionType.valueOf(effectiveSubscriptionStr);
+        }
+        catch (IllegalArgumentException iae) {
+            subscriptionType = SubscriptionType.Exclusive;
+        }
+
+        return subscriptionType;
+    }
+
+    private String getEffectiveConsumerName(String cycleConsumerName) {
+        if ((cycleConsumerName != null) && (!cycleConsumerName.isEmpty())) {
+            return cycleConsumerName;
+        }
+
+        String globalConsumerName = pulsarNBClientConf.getConsumerName();
+        if ((globalConsumerName != null) && (!globalConsumerName.isEmpty())) {
+            return globalConsumerName;
+        }
+
+        return "default-cons";
+    }
+
+    public Consumer<?> getConsumer(String cycleTopicNames,
+                                   String cycleTopicsPattern,
+                                   String cycleSubscriptionName,
+                                   String cycleSubscriptionType,
+                                   String cycleConsumerName) {
+
+        String topicNamesStr = getEffectiveTopicNamesStr(cycleTopicNames);
+        List<String> topicNames = getEffectiveTopicNames(cycleTopicNames);
+        String topicsPatternStr = getEffectiveTopicPatternStr(cycleTopicsPattern);
+        String subscriptionName = getEffectiveSubscriptionName(cycleSubscriptionName);
+        SubscriptionType subscriptionType = getEffectiveSubscriptionType(cycleSubscriptionType);
+        String consumerName = getEffectiveConsumerName(cycleConsumerName);
+
+        String encodedStr = PulsarActivityUtil.encode(
+            consumerName, subscriptionName, topicNamesStr, topicsPatternStr);
+        Consumer<?> consumer = consumers.get(encodedStr);
+
+        if (consumer == null) {
+            PulsarClient pulsarClient = getPulsarClient();
+
+            // Get other possible producer settings that are set at global level
+            Map<String, Object> consumerConf = pulsarNBClientConf.getConsumerConfMap();
+
+            // Explicit topic names will take precedence over topics pattern
+            if ( !topicNames.isEmpty() ) {
+                consumerConf.remove(PulsarActivityUtil.CONSUMER_CONF_KEY.topicsPattern.toString());
+                consumerConf.put(PulsarActivityUtil.CONSUMER_CONF_KEY.topicNames.toString(), topicNames);
+            }
+            else {
+                consumerConf.remove(PulsarActivityUtil.CONSUMER_CONF_KEY.topicNames.toString());
+                if ( !topicsPatternStr.isEmpty() )
+                    consumerConf.put(
+                        PulsarActivityUtil.CONSUMER_CONF_KEY.topicsPattern.toString(),
+                        getEffectiveTopicPattern(cycleTopicsPattern));
+                else {
+                    throw new RuntimeException("\"topicName\" and \"topicsPattern\" can't be empty/invalid at the same time!");
+                }
+            }
+
+            consumerConf.put(PulsarActivityUtil.CONSUMER_CONF_KEY.subscriptionName.toString(), subscriptionName);
+            consumerConf.put(PulsarActivityUtil.CONSUMER_CONF_KEY.subscriptionType.toString(), subscriptionType);
+            consumerConf.put(PulsarActivityUtil.CONSUMER_CONF_KEY.consumerName.toString(), consumerName);
+
+            try {
+                consumer = pulsarClient.newConsumer(pulsarSchema).loadConf(consumerConf).subscribe();
+            }
+            catch (PulsarClientException ple) {
+                ple.printStackTrace();
+                throw new RuntimeException("Unable to create a Pulsar consumer!");
+            }
+
+            consumers.put(encodedStr, consumer);
+        }
+
+        return consumer;
+    }
+}
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarProducerSpace.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarProducerSpace.java
new file mode 100644
index 000000000..d625a165b
--- /dev/null
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarProducerSpace.java
@@ -0,0 +1,81 @@
+package io.nosqlbench.driver.pulsar;
+
+import io.nosqlbench.driver.pulsar.util.PulsarActivityUtil;
+import io.nosqlbench.driver.pulsar.util.PulsarNBClientConf;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+import java.util.Base64;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class PulsarProducerSpace extends  PulsarSpace{
+
+    private final ConcurrentHashMap<String, Producer<?>> producers = new ConcurrentHashMap<>();
+
+    public PulsarProducerSpace(String name, PulsarNBClientConf pulsarClientConf) {
+        super(name, pulsarClientConf);
+    }
+
+    // Producer name is NOT mandatory
+    // - It can be set at either global level or cycle level
+    // - If set at both levels, cycle level setting takes precedence
+    private String getEffectiveProducerName(String cycleProducerName) {
+        if ((cycleProducerName != null) && (!cycleProducerName.isEmpty())) {
+            return cycleProducerName;
+        }
+
+        String globalProducerName = pulsarNBClientConf.getProducerName();
+        if ((globalProducerName != null) && (!globalProducerName.isEmpty())) {
+            return globalProducerName;
+        }
+
+        // Default Producer name when it is not set at either cycle or global level
+        return "default-prod";
+    }
+
+    // Topic name IS mandatory
+    // - It must be set at either global level or cycle level
+    // - If set at both levels, cycle level setting takes precedence
+    private String getEffectiveTopicName(String cycleTopicName) {
+        if ((cycleTopicName != null) && (!cycleTopicName.isEmpty())) {
+            return cycleTopicName;
+        }
+
+        String globalTopicName = pulsarNBClientConf.getProducerTopicName();
+        if ( (globalTopicName == null) || (globalTopicName.isEmpty()) ) {
+            throw new RuntimeException("Topic name must be set at either global level or cycle level!");
+        }
+
+        return globalTopicName;
+    }
+
+    public Producer<?> getProducer(String cycleProducerName, String cycleTopicName) {
+        String producerName = getEffectiveProducerName(cycleProducerName);
+        String topicName = getEffectiveTopicName(cycleTopicName);
+
+        String encodedStr = PulsarActivityUtil.encode(cycleProducerName, cycleTopicName);
+        Producer<?> producer = producers.get(encodedStr);
+
+        if (producer == null) {
+            PulsarClient pulsarClient = getPulsarClient();
+
+            // Get other possible producer settings that are set at global level
+            Map<String, Object> producerConf = pulsarNBClientConf.getProducerConfMap();
+            producerConf.put(PulsarActivityUtil.PRODUCER_CONF_KEY.topicName.toString(), topicName);
+            producerConf.put(PulsarActivityUtil.PRODUCER_CONF_KEY.producerName.toString(), producerName);
+
+            try {
+                producer = pulsarClient.newProducer(pulsarSchema).loadConf(producerConf).create();
+            }
+            catch (PulsarClientException ple) {
+                throw new RuntimeException("Unable to create a Pulsar producer!");
+            }
+
+            producers.put(encodedStr, producer);
+        }
+
+        return producer;
+    }
+}
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarReaderSpace.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarReaderSpace.java
new file mode 100644
index 000000000..3a908b652
--- /dev/null
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarReaderSpace.java
@@ -0,0 +1,15 @@
+package io.nosqlbench.driver.pulsar;
+
+import io.nosqlbench.driver.pulsar.util.PulsarNBClientConf;
+import org.apache.pulsar.client.api.Reader;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+public class PulsarReaderSpace extends PulsarSpace {
+
+    private final ConcurrentHashMap<String, Reader<?>> readers = new ConcurrentHashMap<>();
+
+    public PulsarReaderSpace(String name, PulsarNBClientConf pulsarClientConf) {
+        super(name, pulsarClientConf);
+    }
+}
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarSpace.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarSpace.java
index da754e183..56b50947b 100644
--- a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarSpace.java
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarSpace.java
@@ -17,15 +17,13 @@ import java.util.concurrent.ConcurrentHashMap;
 public class PulsarSpace {
 
     private final static Logger logger = LogManager.getLogger(PulsarSpace.class);
-
-    private final ConcurrentHashMap<String, Producer<?>> producers = new ConcurrentHashMap<>();
     // TODO: add support for other client types: consumer, reader, websocket-producer, managed-ledger, etc.
 
-    private final String name;
-    private final PulsarNBClientConf pulsarNBClientConf;
+    protected final String name;
+    protected final PulsarNBClientConf pulsarNBClientConf;
 
-    private PulsarClient pulsarClient = null;
-    private Schema<?> pulsarSchema = null;
+    protected PulsarClient pulsarClient = null;
+    protected Schema<?> pulsarSchema = null;
 
     public PulsarSpace( String name, PulsarNBClientConf pulsarClientConf ) {
         this.name = name;
@@ -35,7 +33,7 @@ public class PulsarSpace {
         createPulsarSchemaFromConf();
     }
 
-    private void createPulsarClientFromConf() {
+    protected void createPulsarClientFromConf() {
         ClientBuilder clientBuilder = PulsarClient.builder();
 
         String dftSvcUrl = "pulsar://localhost:6650";
@@ -53,11 +51,13 @@ public class PulsarSpace {
         }
     }
 
-    private void createPulsarSchemaFromConf() {
-        String schemaType = pulsarNBClientConf.getSchemaConfValue("schema.type").toString();
+    protected void createPulsarSchemaFromConf() {
+        Object value = pulsarNBClientConf.getSchemaConfValue("schema.type");
+        String schemaType = (value != null) ? value.toString() : "";
 
         if (PulsarActivityUtil.isAvroSchemaTypeStr(schemaType)) {
-            String schemaDefStr = pulsarNBClientConf.getSchemaConfValue("schema.definition").toString();
+            value = pulsarNBClientConf.getSchemaConfValue("schema.definition");
+            String schemaDefStr = (value != null) ? value.toString() : "";
             pulsarSchema = PulsarActivityUtil.getAvroSchema(schemaType, schemaDefStr);
         } else if (PulsarActivityUtil.isPrimitiveSchemaTypeStr(schemaType)) {
             pulsarSchema = PulsarActivityUtil.getPrimitiveTypeSchema((schemaType));
@@ -68,77 +68,8 @@ public class PulsarSpace {
     }
 
     public PulsarClient getPulsarClient() { return pulsarClient; }
-
     public PulsarNBClientConf getPulsarClientConf() {
         return pulsarNBClientConf;
     }
-
     public Schema<?> getPulsarSchema() { return pulsarSchema; }
-
-    // Producer name is NOT mandatory
-    // - It can be set at either global level or cycle level
-    // - If set at both levels, cycle level setting takes precedence
-    private String getEffectiveProducerName(String cycleProducerName) {
-        if ((cycleProducerName != null) && (!cycleProducerName.isEmpty())) {
-            return cycleProducerName;
-        }
-
-        String globalProducerName = pulsarNBClientConf.getProducerName();
-        if ((globalProducerName != null) && (!globalProducerName.isEmpty())) {
-            return globalProducerName;
-        }
-
-        // Default Producer name when it is not set at either cycle or global level
-        return "default";
-    }
-
-    // Topic name is mandatory
-    // - It must be set at either global level or cycle level
-    // - If set at both levels, cycle level setting takes precedence
-    private String getEffectiveTopicName(String cycleTopicName) {
-        if ((cycleTopicName != null) && (!cycleTopicName.isEmpty())) {
-            return cycleTopicName;
-        }
-
-        String globalTopicName = pulsarNBClientConf.getTopicName();
-        if ( (globalTopicName == null) || (globalTopicName.isEmpty()) ) {
-            throw new RuntimeException("Topic name must be set at either global level or cycle level!");
-        }
-
-        return globalTopicName;
-    }
-
-    private Producer createPulsarProducer(String cycleTopicName, String cycleProducerName) {
-        PulsarClient pulsarClient = getPulsarClient();
-        Producer producer = null;
-
-        String producerName = getEffectiveProducerName(cycleProducerName);
-        String topicName = getEffectiveTopicName(cycleTopicName);
-
-        // Get other possible producer settings that are set at global level
-        Map<String, Object> producerConf = pulsarNBClientConf.getProducerConfMap();
-        producerConf.put("topicName", topicName);
-        producerConf.put("producerName", producerName);
-
-        try {
-            producer = pulsarClient.newProducer(pulsarSchema).loadConf(producerConf).create();
-        }
-        catch (PulsarClientException ple) {
-            throw new RuntimeException("Unable to create a client to connect to the Pulsar cluster!");
-        }
-
-        return producer;
-    }
-
-    public Producer<?> getProducer(String cycleProducerName, String cycleTopicName) {
-        String producerName = getEffectiveProducerName(cycleProducerName);
-        Producer producer = producers.get(producerName);
-
-        if (producer == null) {
-            producer = createPulsarProducer(cycleTopicName, cycleProducerName);
-            producers.put(producerName, producer);
-        }
-
-        return producer;
-    }
 }
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarSpaceCache.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarSpaceCache.java
index a9ca5fa77..fe9a64bf7 100644
--- a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarSpaceCache.java
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarSpaceCache.java
@@ -1,5 +1,7 @@
 package io.nosqlbench.driver.pulsar;
 
+import io.nosqlbench.driver.pulsar.util.PulsarActivityUtil;
+
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
@@ -21,8 +23,22 @@ public class PulsarSpaceCache {
     }
 
     public PulsarSpace getPulsarSpace(String name) {
-        PulsarSpace cspace = clientScopes.computeIfAbsent(name, spaceName -> new PulsarSpace(spaceName, activity.getPulsarConf()));
-        return cspace;
+
+        String pulsarClientType = activity.getPulsarConf().getPulsarClientType();
+
+        if (pulsarClientType.equalsIgnoreCase(PulsarActivityUtil.CLIENT_TYPES.PRODUCER.toString())) {
+            return clientScopes.computeIfAbsent(name, spaceName -> new PulsarProducerSpace(spaceName, activity.getPulsarConf()));
+        }
+        else if (pulsarClientType.equalsIgnoreCase(PulsarActivityUtil.CLIENT_TYPES.CONSUMER.toString())) {
+            return clientScopes.computeIfAbsent(name, spaceName -> new PulsarConsumerSpace(spaceName, activity.getPulsarConf()));
+        }
+        else if (pulsarClientType.equalsIgnoreCase(PulsarActivityUtil.CLIENT_TYPES.READER.toString())) {
+            return clientScopes.computeIfAbsent(name, spaceName -> new PulsarReaderSpace(spaceName, activity.getPulsarConf()));
+        }
+        // TODO: add support for websocket-producer and managed-ledger
+        else {
+            throw new RuntimeException("Unsupported Pulsar client type: " + pulsarClientType);
+        }
     }
 
     public PulsarActivity getActivity() {
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarConsumerMapper.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarConsumerMapper.java
index b809dcd07..4a69e6c9f 100644
--- a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarConsumerMapper.java
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarConsumerMapper.java
@@ -2,6 +2,7 @@ package io.nosqlbench.driver.pulsar.ops;
 
 import io.nosqlbench.engine.api.templating.CommandTemplate;
 import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Schema;
 
 import java.util.function.LongFunction;
 
@@ -16,21 +17,21 @@ import java.util.function.LongFunction;
  * For additional parameterization, the command template is also provided.
  */
 public class PulsarConsumerMapper implements LongFunction<PulsarOp> {
-    private final LongFunction<Consumer<?>> consumerFunc;
-    private final LongFunction<String> recvInstructions;
     private final CommandTemplate cmdTpl;
+    private final Schema<?> pulsarSchema;
+    private final LongFunction<Consumer<?>> consumerFunc;
 
-    public PulsarConsumerMapper(LongFunction<Consumer<?>> consumerFunc,
-                                LongFunction<String> recvMsg,
-                                CommandTemplate cmdTpl) {
-        this.consumerFunc = consumerFunc;
-        this.recvInstructions = recvMsg;
+    public PulsarConsumerMapper(CommandTemplate cmdTpl,
+                                Schema<?> pulsarSchema,
+                                LongFunction<Consumer<?>> consumerFunc) {
         this.cmdTpl = cmdTpl;
-        // TODO add schema support
+        this.pulsarSchema = pulsarSchema;
+        this.consumerFunc = consumerFunc;
     }
 
     @Override
     public PulsarOp apply(long value) {
-        return new PulsarConsumerOp((Consumer<byte[]>) consumerFunc.apply(value), recvInstructions.apply(value));
+        Consumer<?> consumer = consumerFunc.apply(value);
+        return new PulsarConsumerOp(consumer, pulsarSchema);
     }
 }
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarConsumerOp.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarConsumerOp.java
index 149931a63..1833772e4 100644
--- a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarConsumerOp.java
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarConsumerOp.java
@@ -1,31 +1,42 @@
 package io.nosqlbench.driver.pulsar.ops;
 
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.PulsarClientException;
+import io.nosqlbench.driver.pulsar.util.AvroUtil;
+import io.nosqlbench.driver.pulsar.util.PulsarActivityUtil;
+import org.apache.pulsar.client.api.*;
+import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.common.schema.SchemaType;
 
 import java.nio.charset.StandardCharsets;
 
 public class PulsarConsumerOp implements PulsarOp {
-    private final Consumer<byte[]> consumer;
-    private final String recvInstructions;
+    private final Consumer<?> consumer;
+    private final Schema<?> pulsarSchema;
 
-    public PulsarConsumerOp(Consumer<byte[]> consumer, String recvInstructions) {
+    public PulsarConsumerOp(Consumer<?> consumer, Schema<?> schema) {
         this.consumer = consumer;
-        this.recvInstructions = recvInstructions;
+        this.pulsarSchema = schema;
     }
 
     @Override
     public void run() {
         try {
-            Message<byte[]> msgbytes = consumer.receive();
-            // TODO: Parameterize the actions taken on a received message
-            // TODO: Properly parameterize all pulsar Op types as with Producer<T> and Consumer<T>
-            String received = new String(msgbytes.getValue(), StandardCharsets.UTF_8);
-            System.out.print("received:" + received);
-            if (!received.endsWith("\n")) {
-                System.out.println("\n");
+            Message<?> message = consumer.receive();
+
+            SchemaType schemaType = pulsarSchema.getSchemaInfo().getType();
+            if (PulsarActivityUtil.isAvroSchemaTypeStr(schemaType.name())) {
+                String avroDefStr = pulsarSchema.getSchemaInfo().getSchemaDefinition();
+                org.apache.avro.generic.GenericRecord avroGenericRecord =
+                    AvroUtil.GetGenericRecord_ApacheAvro(avroDefStr, message.getData());
+
+                System.out.println("msg-key=" + message.getKey() + "  msg-payload=" + avroGenericRecord.toString());
             }
+            else {
+                System.out.println("msg-key=" + message.getKey() + "  msg-payload=" + new String(message.getData()));
+            }
+
+            consumer.acknowledge(message.getMessageId());
+
         } catch (PulsarClientException e) {
             throw new RuntimeException(e);
         }
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarProducerMapper.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarProducerMapper.java
index 30231d1ae..cebc8287c 100644
--- a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarProducerMapper.java
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarProducerMapper.java
@@ -18,23 +18,22 @@ import java.util.function.LongFunction;
  * For additional parameterization, the command template is also provided.
  */
 public class PulsarProducerMapper implements LongFunction<PulsarOp> {
+    private final CommandTemplate cmdTpl;
+    private final Schema<?> pulsarSchema;
     private final LongFunction<Producer<?>> producerFunc;
     private final LongFunction<String> keyFunc;
     private final LongFunction<String> payloadFunc;
-    private final Schema pulsarSchema;
-    private final CommandTemplate cmdTpl;
 
-    public PulsarProducerMapper(
-        LongFunction<Producer<?>> producerFunc,
-        LongFunction<String> keyFunc,
-        LongFunction<String> payloadFunc,
-        Schema pulsarSchema,
-        CommandTemplate cmdTpl) {
+    public PulsarProducerMapper(CommandTemplate cmdTpl,
+                                Schema<?> pulsarSchema,
+                                LongFunction<Producer<?>> producerFunc,
+                                LongFunction<String> keyFunc,
+                                LongFunction<String> payloadFunc) {
+        this.cmdTpl = cmdTpl;
+        this.pulsarSchema = pulsarSchema;
         this.producerFunc = producerFunc;
         this.keyFunc = keyFunc;
         this.payloadFunc = payloadFunc;
-        this.pulsarSchema = pulsarSchema;
-        this.cmdTpl = cmdTpl;
     }
 
     @Override
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarReaderMapper.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarReaderMapper.java
new file mode 100644
index 000000000..7ae971a7e
--- /dev/null
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarReaderMapper.java
@@ -0,0 +1,27 @@
+package io.nosqlbench.driver.pulsar.ops;
+
+import io.nosqlbench.engine.api.templating.CommandTemplate;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.Schema;
+
+import java.util.function.LongFunction;
+
+public class PulsarReaderMapper implements LongFunction<PulsarOp> {
+    private final CommandTemplate cmdTpl;
+    private final Schema<?> pulsarSchema;
+    private final LongFunction<Reader<?>> readerFunc;
+
+    public PulsarReaderMapper(CommandTemplate cmdTpl,
+                              Schema<?> pulsarSchema,
+                              LongFunction<Reader<?>> readerFunc) {
+        this.cmdTpl = cmdTpl;
+        this.pulsarSchema = pulsarSchema;
+        this.readerFunc = readerFunc;
+    }
+
+    @Override
+    public PulsarOp apply(long value) {
+        Reader<?> reader = readerFunc.apply(value);
+        return new PulsarReaderOp(reader, pulsarSchema);
+    }
+}
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarReaderOp.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarReaderOp.java
new file mode 100644
index 000000000..d3d7bd94f
--- /dev/null
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarReaderOp.java
@@ -0,0 +1,19 @@
+package io.nosqlbench.driver.pulsar.ops;
+
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.Schema;
+
+public class PulsarReaderOp implements PulsarOp {
+    private final Reader<?> reader;
+    private final Schema<?> pulsarSchema;
+
+    public PulsarReaderOp(Reader<?> reader, Schema<?> schema) {
+        this.reader = reader;
+        this.pulsarSchema = schema;
+    }
+
+    @Override
+    public void run() {
+        //TODO: to be added
+    }
+}
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/ReadyPulsarOp.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/ReadyPulsarOp.java
index 82272150f..8add08167 100644
--- a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/ReadyPulsarOp.java
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/ReadyPulsarOp.java
@@ -1,13 +1,13 @@
 package io.nosqlbench.driver.pulsar.ops;
 
+import io.nosqlbench.driver.pulsar.*;
 import io.nosqlbench.driver.pulsar.util.PulsarActivityUtil;
-import io.nosqlbench.driver.pulsar.PulsarSpace;
-import io.nosqlbench.driver.pulsar.PulsarSpaceCache;
-import io.nosqlbench.driver.pulsar.util.PulsarNBClientConf;
 import io.nosqlbench.engine.api.activityconfig.yaml.OpTemplate;
 import io.nosqlbench.engine.api.scoping.ScopedSupplier;
 import io.nosqlbench.engine.api.templating.CommandTemplate;
 import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Reader;
 import org.apache.pulsar.client.api.Schema;
 
 import java.util.function.LongFunction;
@@ -15,22 +15,23 @@ import java.util.function.Supplier;
 
 public class ReadyPulsarOp implements LongFunction<PulsarOp> {
 
+    private final OpTemplate opTpl;
     private final CommandTemplate cmdTpl;
     private final PulsarSpace clientSpace;
     private final LongFunction<PulsarOp> opFunc;
-    private final PulsarSpaceCache pcache;
-    private final Schema pulsarSchema;
+    private final Schema<?> pulsarSchema;
 
     // TODO: Add docs for the command template with respect to the OpTemplate
 
     public ReadyPulsarOp(OpTemplate opTemplate, PulsarSpaceCache pcache) {
         // TODO: Consider parsing map structures into equivalent binding representation
+        this.opTpl = opTemplate;
         this.cmdTpl = new CommandTemplate(opTemplate);
+
         if (cmdTpl.isDynamic("op_scope")) {
             throw new RuntimeException("op_scope must be static");
         }
 
-        this.pcache = pcache;
         // TODO: At the moment, only supports static "client"
         if (cmdTpl.containsKey("client")) {
             if (cmdTpl.isDynamic("client")) {
@@ -52,7 +53,30 @@ public class ReadyPulsarOp implements LongFunction<PulsarOp> {
     }
 
     private LongFunction<PulsarOp> resolve() {
+        String clientType = clientSpace.getPulsarClientConf().getPulsarClientType();
 
+        // TODO: Complete implementation for reader, websocket-producer and managed-ledger
+        if ( clientType.equalsIgnoreCase(PulsarActivityUtil.CLIENT_TYPES.PRODUCER.toString()) ) {
+            assert clientSpace instanceof PulsarProducerSpace;
+            return resolveProducer((PulsarProducerSpace) clientSpace, cmdTpl);
+        } else if ( clientType.equalsIgnoreCase(PulsarActivityUtil.CLIENT_TYPES.CONSUMER.toString()) ) {
+            assert clientSpace instanceof PulsarConsumerSpace;
+            return resolveConsumer((PulsarConsumerSpace)clientSpace, cmdTpl); /*
+        } else if ( clientType.equalsIgnoreCase(PulsarActivityUtil.CLIENT_TYPES.READER.toString()) ) {
+            assert clientSpace instanceof PulsarReaderSpace;
+            return resolveReader((PulsarReaderSpace)clientSpace, cmdTpl);
+        } else if ( clientType.equalsIgnoreCase(PulsarActivityUtil.CLIENT_TYPES.WSOKT_PRODUCER.toString()) ) {
+        } else if ( clientType.equalsIgnoreCase(PulsarActivityUtil.CLIENT_TYPES.MANAGED_LEDGER.toString()) ) {
+        */
+        } else {
+            throw new RuntimeException("Unsupported Pulsar client: " + clientType);
+        }
+    }
+
+    private LongFunction<PulsarOp> resolveProducer(
+        PulsarProducerSpace clientSpace,
+        CommandTemplate cmdTpl
+    ) {
         if (cmdTpl.containsKey("topic_url")) {
             throw new RuntimeException("topic_url is not valid. Perhaps you mean topic_uri ?");
         }
@@ -99,31 +123,8 @@ public class ReadyPulsarOp implements LongFunction<PulsarOp> {
             topic_uri_func = (l) -> null;
         }
 
-        assert (clientSpace != null);
-        String clientType = clientSpace.getPulsarClientConf().getPulsarClientType();
-
-        // TODO: At the moment, only implements "Producer" functionality; add implementation for others later!
-        if ( clientType.equalsIgnoreCase(PulsarActivityUtil.CLIENT_TYPES.PRODUCER.toString()) ) {
-            return resolveProducer(clientSpace, cmdTpl, cycle_producer_name_func, topic_uri_func);/*
-        } else if ( msgOperation.equalsIgnoreCase(PulsarActivityUtil.MSGOP_TYPES.CONSUMER.toString()) ) {
-            return resolveConsumer(spaceFunc, cmdTpl, topic_uri_func);
-        } else if ( msgOperation.equalsIgnoreCase(PulsarOpUtil.MSGOP_TYPES.READER.toString()) ) {
-        } else if ( msgOperation.equalsIgnoreCase(PulsarOpUtil.MSGOP_TYPES.WSOKT_PRODUCER.toString()) ) {
-        } else if ( msgOperation.equalsIgnoreCase(PulsarOpUtil.MSGOP_TYPES.MANAGED_LEDGER.toString()) ) {
-        */
-        } else {
-            throw new RuntimeException("Unsupported Pulsar message operation type.");
-        }
-    }
-
-    private LongFunction<PulsarOp> resolveProducer(
-        PulsarSpace pulsarSpace,
-        CommandTemplate cmdTpl,
-        LongFunction<String> cycle_producer_name_func,
-        LongFunction<String> topic_uri_func
-    ) {
         LongFunction<Producer<?>> producerFunc =
-            (l) -> pulsarSpace.getProducer(cycle_producer_name_func.apply(l), topic_uri_func.apply(l));
+            (l) -> clientSpace.getProducer(cycle_producer_name_func.apply(l), topic_uri_func.apply(l));
 
         LongFunction<String> keyFunc;
         if (cmdTpl.isStatic("msg-key")) {
@@ -147,12 +148,80 @@ public class ReadyPulsarOp implements LongFunction<PulsarOp> {
             throw new RuntimeException("\"msg-value\" field must be specified!");
         }
 
-        return new PulsarProducerMapper(producerFunc, keyFunc, valueFunc, pulsarSchema, cmdTpl);
+        return new PulsarProducerMapper(cmdTpl, pulsarSchema, producerFunc, keyFunc, valueFunc);
+    }
+
+    private LongFunction<PulsarOp> resolveConsumer(
+        PulsarConsumerSpace clientSpace,
+        CommandTemplate cmdTpl
+    ) {
+        LongFunction<String> topic_names_func;
+        if (cmdTpl.isStatic("topic-names")) {
+            topic_names_func = (l) -> cmdTpl.getStatic("topic-names");
+        } else if (cmdTpl.isDynamic("topic-names")) {
+            topic_names_func = (l) -> cmdTpl.getDynamic("topic-names", l);
+        } else {
+            topic_names_func = (l) -> null;
+        }
+
+        LongFunction<String> topics_pattern_func;
+        if (cmdTpl.isStatic("topics-pattern")) {
+            topics_pattern_func = (l) -> cmdTpl.getStatic("topics-pattern");
+        } else if (cmdTpl.isDynamic("topics-pattern")) {
+            topics_pattern_func = (l) -> cmdTpl.getDynamic("topics-pattern", l);
+        } else {
+            topics_pattern_func = (l) -> null;
+        }
+
+        LongFunction<String> subscription_name_func;
+        if (cmdTpl.isStatic("subscription-name")) {
+            subscription_name_func = (l) -> cmdTpl.getStatic("subscription-name");
+        } else if (cmdTpl.isDynamic("subscription-name")) {
+            subscription_name_func = (l) -> cmdTpl.getDynamic("subscription-name", l);
+        } else {
+            subscription_name_func = (l) -> null;
+        }
+
+        LongFunction<String> subscription_type_func;
+        if (cmdTpl.isStatic("subscription-type")) {
+            subscription_type_func = (l) -> cmdTpl.getStatic("subscription-type");
+        } else if (cmdTpl.isDynamic("subscription-type")) {
+            subscription_type_func = (l) -> cmdTpl.getDynamic("subscription-type", l);
+        } else {
+            subscription_type_func = (l) -> null;
+        }
+
+        LongFunction<String> consumer_name_func;
+        if (cmdTpl.isStatic("consumer-name")) {
+            consumer_name_func = (l) -> cmdTpl.getStatic("consumer-name");
+        } else if (cmdTpl.isDynamic("consumer-name")) {
+            consumer_name_func = (l) -> cmdTpl.getDynamic("consumer-name", l);
+        } else {
+            consumer_name_func = (l) -> null;
+        }
+
+        LongFunction<Consumer<?>> consumerFunc = (l) ->
+            clientSpace.getConsumer(
+                topic_names_func.apply(l),
+                topics_pattern_func.apply(l),
+                subscription_name_func.apply(l),
+                subscription_type_func.apply(l),
+                consumer_name_func.apply(l)
+            );
+
+        return new PulsarConsumerMapper(cmdTpl, pulsarSchema, consumerFunc);
+    }
+
+    private LongFunction<PulsarOp> resolveReader(
+        PulsarReaderSpace pulsarSpace,
+        CommandTemplate cmdTpl
+    ) {
+        //TODO: to be completed
+        return null;
     }
 
     @Override
     public PulsarOp apply(long value) {
-        PulsarOp op = opFunc.apply(value);
-        return op;
+        return opFunc.apply(value);
     }
 }
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/AvroUtil.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/AvroUtil.java
index cf7537f5c..2a4f1f75e 100644
--- a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/AvroUtil.java
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/AvroUtil.java
@@ -2,6 +2,7 @@ package io.nosqlbench.driver.pulsar.util;
 
 import org.apache.avro.io.DecoderFactory;
 import org.apache.avro.io.JsonDecoder;
+import org.apache.avro.io.BinaryDecoder;
 import org.apache.pulsar.client.api.schema.Field;
 import org.apache.pulsar.client.api.schema.GenericRecord;
 import org.apache.pulsar.client.api.schema.GenericRecordBuilder;
@@ -25,6 +26,7 @@ public class AvroUtil {
 
             org.apache.avro.generic.GenericDatumReader<org.apache.avro.generic.GenericData.Record> reader;
             reader = new org.apache.avro.generic.GenericDatumReader<>(schema);
+
             JsonDecoder decoder = DecoderFactory.get().jsonDecoder(schema, jsonData);
 
             record = reader.read(null, decoder);
@@ -35,6 +37,25 @@ public class AvroUtil {
 
         return record;
     }
+    public static org.apache.avro.generic.GenericRecord GetGenericRecord_ApacheAvro(String avroSchemDef, byte[] bytesData)  {
+        org.apache.avro.generic.GenericRecord record = null;
+
+        try {
+            org.apache.avro.Schema schema = GetSchema_ApacheAvro(avroSchemDef);
+
+            org.apache.avro.generic.GenericDatumReader<org.apache.avro.generic.GenericData.Record> reader;
+            reader = new org.apache.avro.generic.GenericDatumReader<>(schema);
+
+            BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(bytesData, null);
+
+            record = reader.read(null, decoder);
+        }
+        catch (IOException ioe) {
+            ioe.printStackTrace();
+        }
+
+        return record;
+    }
 
     public static GenericAvroSchema GetSchema_PulsarAvro(String schemaName, String avroSchemDef) {
         SchemaInfo schemaInfo = SchemaInfo.builder()
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/PulsarActivityUtil.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/PulsarActivityUtil.java
index 6247a566f..ffeacb61b 100644
--- a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/PulsarActivityUtil.java
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/PulsarActivityUtil.java
@@ -15,6 +15,7 @@ import java.nio.file.InvalidPathException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Arrays;
+import java.util.Base64;
 import java.util.HashMap;
 
 public class PulsarActivityUtil {
@@ -31,7 +32,7 @@ public class PulsarActivityUtil {
         ;
 
         public final String label;
-        private CLIENT_TYPES(String label) {
+        CLIENT_TYPES(String label) {
             this.label = label;
         }
     }
@@ -48,7 +49,7 @@ public class PulsarActivityUtil {
         ;
 
         public final String label;
-        private PERSISTENT_TYPES(String label) {
+        PERSISTENT_TYPES(String label) {
             this.label = label;
         }
     }
@@ -59,6 +60,7 @@ public class PulsarActivityUtil {
 
     ///////
     // Valid Pulsar client configuration (activity-level settings)
+    // - https://pulsar.apache.org/docs/en/client-libraries-java/#client
     public enum CLNT_CONF_KEY {
         serviceUrl("serviceUrl"),
         authPulginClassName("authPluginClassName"),
@@ -83,7 +85,7 @@ public class PulsarActivityUtil {
         ;
 
         public final String label;
-        private CLNT_CONF_KEY(String label) {
+        CLNT_CONF_KEY(String label) {
             this.label = label;
         }
     }
@@ -93,12 +95,10 @@ public class PulsarActivityUtil {
 
     ///////
     // Valid producer configuration (activity-level settings)
+    // - https://pulsar.apache.org/docs/en/client-libraries-java/#configure-producer
     public enum PRODUCER_CONF_KEY {
-        // NOTE:
-        //   For "topicName" and "producerName", they're ignore at activity-level.
-        //   Instead, op-level settings are respected
-        // topicName("topicName"),
-        // producerName("producerName"),
+        topicName("topicName"),
+        producerName("producerName"),
         sendTimeoutMs("sendTimeoutMs"),
         blockIfQueueFull("blockIfQueueFull"),
         maxPendingMessages("maxPendingMessages"),
@@ -113,7 +113,7 @@ public class PulsarActivityUtil {
         ;
 
         public final String label;
-        private PRODUCER_CONF_KEY(String label) {
+        PRODUCER_CONF_KEY(String label) {
             this.label = label;
         }
     }
@@ -123,27 +123,63 @@ public class PulsarActivityUtil {
 
     ///////
     // Valid consumer configuration (activity-level settings)
-    // TODO: to be added
+    // - https://pulsar.apache.org/docs/en/client-libraries-java/#consumer
     public enum CONSUMER_CONF_KEY {
+        topicNames("topicNames"),
+        topicsPattern("topicsPattern"),
+        subscriptionName("subscriptionName"),
+        subscriptionType("subscriptionType"),
+        receiverQueueSize("receiverQueueSize"),
+        acknowledgementsGroupTimeMicros("acknowledgementsGroupTimeMicros"),
+        negativeAckRedeliveryDelayMicros("negativeAckRedeliveryDelayMicros"),
+        maxTotalReceiverQueueSizeAcrossPartitions("maxTotalReceiverQueueSizeAcrossPartitions"),
+        consumerName("consumerName"),
+        ackTimeoutMillis("ackTimeoutMillis"),
+        tickDurationMillis("tickDurationMillis"),
+        priorityLevel("priorityLevel"),
+        cryptoFailureAction("cryptoFailureAction"),
+        properties("properties"),
+        readCompacted("readCompacted"),
+        subscriptionInitialPosition("subscriptionInitialPosition"),
+        patternAutoDiscoveryPeriod("patternAutoDiscoveryPeriod"),
+        regexSubscriptionMode("regexSubscriptionMode"),
+        deadLetterPolicy("deadLetterPolicy"),
+        autoUpdatePartitions("autoUpdatePartitions"),
+        replicateSubscriptionState("replicateSubscriptionState")
         ;
 
         public final String label;
-        private CONSUMER_CONF_KEY(String label) {
+        CONSUMER_CONF_KEY(String label) {
             this.label = label;
         }
     }
+    public static boolean isValidConsumerConfItem(String item) {
+        return Arrays.stream(CONSUMER_CONF_KEY.values()).anyMatch((t) -> t.name().equals(item.toLowerCase()));
+    }
 
     ///////
     // Valid reader configuration (activity-level settings)
-    // TODO: to be added
+    // - https://pulsar.apache.org/docs/en/client-libraries-java/#reader
     public enum READER_CONF_KEY {
+        topicName("topicName"),
+        receiverQueueSize("receiverQueueSize"),
+        readerListener("readerListener"),
+        readerName("readerName"),
+        subscriptionRolePrefix("subscriptionRolePrefix"),
+        cryptoKeyReader("cryptoKeyReader"),
+        cryptoFailureAction("cryptoFailureAction"),
+        readCompacted("readCompacted"),
+        resetIncludeHead("resetIncludeHead")
         ;
 
         public final String label;
-        private READER_CONF_KEY(String label) {
+        READER_CONF_KEY(String label) {
             this.label = label;
         }
     }
+    public static boolean isValidReaderConfItem(String item) {
+        return Arrays.stream(READER_CONF_KEY.values()).anyMatch((t) -> t.name().equals(item.toLowerCase()));
+    }
 
     ///////
     // Valid websocket-producer configuration (activity-level settings)
@@ -152,7 +188,7 @@ public class PulsarActivityUtil {
         ;
 
         public final String label;
-        private WEBSKT_PRODUCER_CONF_KEY(String label) {
+        WEBSKT_PRODUCER_CONF_KEY(String label) {
             this.label = label;
         }
     }
@@ -164,7 +200,7 @@ public class PulsarActivityUtil {
         ;
 
         public final String label;
-        private MANAGED_LEDGER_CONF_KEY(String label) {
+        MANAGED_LEDGER_CONF_KEY(String label) {
             this.label = label;
         }
     }
@@ -179,21 +215,21 @@ public class PulsarActivityUtil {
             typeStr = "BYTES";
         }
 
-        if ( typeStr.toUpperCase().equals("BOOLEAN") || typeStr.toUpperCase().equals("INT8") ||
-             typeStr.toUpperCase().equals("INT16") || typeStr.toUpperCase().equals("INT32") ||
-             typeStr.toUpperCase().equals("INT64") || typeStr.toUpperCase().equals("FLOAT") ||
-             typeStr.toUpperCase().equals("DOUBLE") || typeStr.toUpperCase().equals("BYTES") ||
-             typeStr.toUpperCase().equals("DATE") || typeStr.toUpperCase().equals("TIME") ||
-             typeStr.toUpperCase().equals("TIMESTAMP") || typeStr.toUpperCase().equals("INSTANT") ||
-             typeStr.toUpperCase().equals("LOCAL_DATE") || typeStr.toUpperCase().equals("LOCAL_TIME") ||
-             typeStr.toUpperCase().equals("LOCAL_DATE_TIME") ) {
+        if ( typeStr.equalsIgnoreCase("BOOLEAN") || typeStr.equalsIgnoreCase("INT8") ||
+             typeStr.equalsIgnoreCase("INT16") || typeStr.equalsIgnoreCase("INT32") ||
+             typeStr.equalsIgnoreCase("INT64") || typeStr.equalsIgnoreCase("FLOAT") ||
+             typeStr.equalsIgnoreCase("DOUBLE") || typeStr.equalsIgnoreCase("BYTES") ||
+             typeStr.equalsIgnoreCase("DATE") || typeStr.equalsIgnoreCase("TIME") ||
+             typeStr.equalsIgnoreCase("TIMESTAMP") || typeStr.equalsIgnoreCase("INSTANT") ||
+             typeStr.equalsIgnoreCase("LOCAL_DATE") || typeStr.equalsIgnoreCase("LOCAL_TIME") ||
+             typeStr.equalsIgnoreCase("LOCAL_DATE_TIME") ) {
             isPrimitive = true;
         }
 
         return isPrimitive;
     }
-    public static Schema getPrimitiveTypeSchema(String typeStr) {
-        Schema schema = null;
+    public static Schema<?> getPrimitiveTypeSchema(String typeStr) {
+        Schema<?> schema;
 
         switch (typeStr.toUpperCase()) {
             case "BOOLEAN":
@@ -240,7 +276,7 @@ public class PulsarActivityUtil {
                 break;
             // Use BYTES as the default schema type if the type string is not specified
             case "":
-            case "BTYES":
+            case "BYTES":
                 schema = Schema.BYTES;
                 break;
             // Report an error if non-valid, non-empty schema type string is provided
@@ -255,15 +291,15 @@ public class PulsarActivityUtil {
     // Complex strut type: Avro or Json
     public static boolean isAvroSchemaTypeStr(String typeStr) {
         boolean isAvroType = false;
-        if ( typeStr.toUpperCase().equals("AVRO") ) {
+        if ( typeStr.equalsIgnoreCase("AVRO") ) {
             isAvroType = true;
         }
         return isAvroType;
     }
-    public static Schema getAvroSchema(String typeStr, String definitionStr) {
+    public static Schema<?> getAvroSchema(String typeStr, String definitionStr) {
         String schemaDefinitionStr = definitionStr;
         String filePrefix = "file://";
-        Schema schema = null;
+        Schema<?> schema;
 
         // Check if payloadStr points to a file (e.g. "file:///path/to/a/file")
         if (isAvroSchemaTypeStr(typeStr)) {
@@ -278,8 +314,6 @@ public class PulsarActivityUtil {
                 }
             }
 
-            System.out.println(schemaDefinitionStr);
-
             SchemaInfo schemaInfo = SchemaInfo.builder()
                 .schema(schemaDefinitionStr.getBytes(StandardCharsets.UTF_8))
                 .type(SchemaType.AVRO)
@@ -296,5 +330,16 @@ public class PulsarActivityUtil {
 
         return schema;
     }
+
+    public static String encode(String... strings) {
+        StringBuilder stringBuilder = new StringBuilder();
+
+        for (String str : strings) {
+            if ((str != null) && !str.isEmpty())
+                stringBuilder.append(str).append("::");
+        }
+
+        return Base64.getEncoder().encodeToString(stringBuilder.toString().getBytes());
+    }
 }
 
diff --git a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/PulsarNBClientConf.java b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/PulsarNBClientConf.java
index ec8c4a0d2..e5b6a58c6 100644
--- a/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/PulsarNBClientConf.java
+++ b/driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/util/PulsarNBClientConf.java
@@ -26,11 +26,15 @@ public class PulsarNBClientConf {
     private static final String SCHEMA_CONF_PREFIX = "schema";
     private static final String CLIENT_CONF_PREFIX = "client";
     private static final String PRODUCER_CONF_PREFIX = "producer";
+    private static final String CONSUMER_CONF_PREFIX = "consumer";
+    private static final String READER_CONF_PREFIX = "reader";
     private HashMap<String, Object> driverConfMap = new HashMap<>();
     private HashMap<String, Object> schemaConfMap = new HashMap<>();
     private HashMap<String, Object> clientConfMap = new HashMap<>();
     private HashMap<String, Object> producerConfMap = new HashMap<>();
-    // TODO: add support for other operation types: consumer, reader, websocket-producer, managed-ledger
+    private HashMap<String, Object> consumerConfMap = new HashMap<>();
+    private HashMap<String, Object> readerConfMap = new HashMap<>();
+    // TODO: add support for other operation types: websocket-producer, managed-ledger
 
     public PulsarNBClientConf(String fileName) {
         File file = new File(fileName);
@@ -51,25 +55,49 @@ public class PulsarNBClientConf {
             // Get driver specific configuration settings
             for (Iterator<String> it = config.getKeys(DRIVER_CONF_PREFIX); it.hasNext(); ) {
                 String confKey = it.next();
-                driverConfMap.put(confKey.substring(DRIVER_CONF_PREFIX.length()+1), config.getProperty(confKey));
+                String confVal = config.getProperty(confKey).toString();
+                if ( (confVal != null) && !confVal.isEmpty() )
+                    driverConfMap.put(confKey.substring(DRIVER_CONF_PREFIX.length()+1), config.getProperty(confKey));
             }
 
             // Get schema specific configuration settings
             for (Iterator<String> it = config.getKeys(SCHEMA_CONF_PREFIX); it.hasNext(); ) {
                 String confKey = it.next();
-                schemaConfMap.put(confKey.substring(SCHEMA_CONF_PREFIX.length()+1), config.getProperty(confKey));
+                String confVal = config.getProperty(confKey).toString();
+                if ( (confVal != null) && !confVal.isEmpty() )
+                    schemaConfMap.put(confKey.substring(SCHEMA_CONF_PREFIX.length()+1), config.getProperty(confKey));
             }
 
             // Get client connection specific configuration settings
             for (Iterator<String> it = config.getKeys(CLIENT_CONF_PREFIX); it.hasNext(); ) {
                 String confKey = it.next();
-                clientConfMap.put(confKey.substring(CLIENT_CONF_PREFIX.length()+1), config.getProperty(confKey));
+                String confVal = config.getProperty(confKey).toString();
+                if ( (confVal != null) && !confVal.isEmpty() )
+                    clientConfMap.put(confKey.substring(CLIENT_CONF_PREFIX.length()+1), config.getProperty(confKey));
             }
 
             // Get producer specific configuration settings
             for (Iterator<String> it = config.getKeys(PRODUCER_CONF_PREFIX); it.hasNext(); ) {
                 String confKey = it.next();
-                producerConfMap.put(confKey.substring(PRODUCER_CONF_PREFIX.length()+1), config.getProperty(confKey));
+                String confVal = config.getProperty(confKey).toString();
+                if ( (confVal != null) && !confVal.isEmpty() )
+                    producerConfMap.put(confKey.substring(PRODUCER_CONF_PREFIX.length()+1), config.getProperty(confKey));
+            }
+
+            // Get producer specific configuration settings
+            for (Iterator<String> it = config.getKeys(CONSUMER_CONF_PREFIX); it.hasNext(); ) {
+                String confKey = it.next();
+                String confVal = config.getProperty(confKey).toString();
+                if ( (confVal != null) && !confVal.isEmpty() )
+                    consumerConfMap.put(confKey.substring(CONSUMER_CONF_PREFIX.length()+1), config.getProperty(confKey));
+            }
+
+            // Get producer specific configuration settings
+            for (Iterator<String> it = config.getKeys(READER_CONF_PREFIX); it.hasNext(); ) {
+                String confKey = it.next();
+                String confVal = config.getProperty(confKey).toString();
+                if ( (confVal != null) && !confVal.isEmpty() )
+                    readerConfMap.put(confKey.substring(READER_CONF_PREFIX.length()+1), config.getProperty(confKey));
             }
         }
         catch (IOException ioe) {
@@ -82,6 +110,8 @@ public class PulsarNBClientConf {
         }
     }
 
+
+    //////////////////
     // Get NB Driver related config
     public Map<String, Object> getDriverConfMap() {
         return this.driverConfMap;
@@ -104,7 +134,19 @@ public class PulsarNBClientConf {
         else
             driverConfMap.put(key, value);
     }
+    // other driver helper functions ...
+    public String getPulsarClientType() {
+        Object confValue = getDriverConfValue("driver.client-type");
 
+        // If not explicitly specifying Pulsar client type, "producer" is the default type
+        if (confValue == null)
+            return PulsarActivityUtil.CLIENT_TYPES.PRODUCER.toString();
+        else
+            return confValue.toString();
+    }
+
+
+    //////////////////
     // Get Schema related config
     public Map<String, Object> getSchemaConfMap() {
         return this.schemaConfMap;
@@ -128,6 +170,8 @@ public class PulsarNBClientConf {
             schemaConfMap.put(key, value);
     }
 
+
+    //////////////////
     // Get Pulsar client related config
     public Map<String, Object> getClientConfMap() {
         return this.clientConfMap;
@@ -151,6 +195,8 @@ public class PulsarNBClientConf {
             clientConfMap.put(key, value);
     }
 
+
+    //////////////////
     // Get Pulsar producer related config
     public Map<String, Object> getProducerConfMap() {
         return this.producerConfMap;
@@ -173,34 +219,106 @@ public class PulsarNBClientConf {
         else
             producerConfMap.put(key, value);
     }
-
-    public String getPulsarClientType() {
-        Object confValue = getDriverConfValue("driver.client-type");
-
-        // If not explicitly specifying Pulsar client type, "producer" is the default type
-        if (confValue == null)
-            return PulsarActivityUtil.CLIENT_TYPES.PRODUCER.toString();
-        else
-            return confValue.toString();
-    }
-
+    // other producer helper functions ...
     public String getProducerName() {
         Object confValue = getProducerConfValue("producer.producerName");
-
-        // If not explicitly specifying Pulsar client type, "producer" is the default type
         if (confValue == null)
             return "";
         else
             return confValue.toString();
     }
-
-    public String getTopicName() {
+    public String getProducerTopicName() {
         Object confValue = getProducerConfValue("producer.topicName");
-
-        // If not explicitly specifying Pulsar client type, "producer" is the default type
         if (confValue == null)
             return "";
         else
             return confValue.toString();
     }
+
+
+    //////////////////
+    // Get Pulsar consumer related config
+    public Map<String, Object> getConsumerConfMap() {
+        return this.consumerConfMap;
+    }
+    public boolean hasConsumerConfKey(String key) {
+        if (key.contains(CONSUMER_CONF_PREFIX))
+            return consumerConfMap.containsKey(key.substring(CONSUMER_CONF_PREFIX.length()+1));
+        else
+            return consumerConfMap.containsKey(key);
+    }
+    public Object getConsumerConfValue(String key) {
+        if (key.contains(CONSUMER_CONF_PREFIX))
+            return consumerConfMap.get(key.substring(CONSUMER_CONF_PREFIX.length()+1));
+        else
+            return consumerConfMap.get(key);
+    }
+    public void setConsumerConfValue(String key, Object value) {
+        if (key.contains(CONSUMER_CONF_PREFIX))
+            consumerConfMap.put(key.substring(CONSUMER_CONF_PREFIX.length()+1), value);
+        else
+            consumerConfMap.put(key, value);
+    }
+    // Other consumer helper functions ...
+    public String getConsumerTopicNames() {
+        Object confValue = getConsumerConfValue("consumer.topicNames");
+        if (confValue == null)
+            return "";
+        else
+            return confValue.toString();
+    }
+    public String getConsumerTopicPattern() {
+        Object confValue = getConsumerConfValue("consumer.topicsPattern");
+        if (confValue == null)
+            return "";
+        else
+            return confValue.toString();
+    }
+    public String getConsumerSubscriptionName() {
+        Object confValue = getConsumerConfValue("consumer.subscriptionName");
+        if (confValue == null)
+            return "";
+        else
+            return confValue.toString();
+    }
+    public String getConsumerSubscriptionType() {
+        Object confValue = getConsumerConfValue("consumer.subscriptionType");
+        if (confValue == null)
+            return "";
+        else
+            return confValue.toString();
+    }
+    public String getConsumerName() {
+        Object confValue = getConsumerConfValue("consumer.consumerName");
+        if (confValue == null)
+            return "";
+        else
+            return confValue.toString();
+    }
+
+
+    //////////////////
+    // Get Pulsar reader related config
+    public Map<String, Object> getReaderConfMap() {
+        return this.readerConfMap;
+    }
+    public boolean hasReaderConfKey(String key) {
+        if (key.contains(READER_CONF_PREFIX))
+            return readerConfMap.containsKey(key.substring(READER_CONF_PREFIX.length()+1));
+        else
+            return readerConfMap.containsKey(key);
+    }
+    public Object getReaderConfValue(String key) {
+        if (key.contains(READER_CONF_PREFIX))
+            return readerConfMap.get(key.substring(READER_CONF_PREFIX.length()+1));
+        else
+            return readerConfMap.get(key);
+    }
+    public void setReaderConfValue(String key, Object value) {
+        if (key.contains(READER_CONF_PREFIX))
+            readerConfMap.put(key.substring(READER_CONF_PREFIX.length()+1), value);
+        else
+            readerConfMap.put(key, value);
+    }
+    // Other consumer helper functions ...
 }
diff --git a/driver-pulsar/src/main/resources/activities/config.properties b/driver-pulsar/src/main/resources/activities/config.properties
index 69666467b..925f03477 100644
--- a/driver-pulsar/src/main/resources/activities/config.properties
+++ b/driver-pulsar/src/main/resources/activities/config.properties
@@ -1,9 +1,9 @@
 ### NB Pulsar driver related configuration - driver.xxx
-driver.client-type = producer
-# TODO: At the moment, only one producer is publishing messages to one single topic through NB
-# TODO  - consider allowing multiple producers to publish to the same topic
-# TODO  - but this relies on the NB cycle to be able to run Pulsar client asynchronously!
+driver.client-type = consumer
 driver.num-workers = 1
+# TODO: functionalities to be completed
+driver.sync-mode = sync
+driver.msg-recv-ouput = console
 
 
 ### Schema related configurations - schema.xxx
@@ -13,17 +13,16 @@ driver.num-workers = 1
 # - strut (complex type) (https://pulsar.apache.org/docs/en/schema-understand/#struct)
 #   avro, json, protobuf
 #
-# TODO: as a starting point, only supports:
-# TODO: 1) primitive types, including bytearray (byte[]) which is default, for messages without schema
-# TODO: 2) Avro for messages with schema
-schema.type = avro
+# TODO: as a starting point, only supports the following types
+#       1) primitive types, including bytearray (byte[]) which is default, for messages without schema
+#       2) Avro for messages with schema
+schema.type =
 schema.definition = file:///Users/yabinmeng/DataStax/nosqlbench/driver-pulsar/src/main/resources/activities/iot-example.avsc
 
 
 ### Pulsar client related configurations - client.xxx
 # http://pulsar.apache.org/docs/en/client-libraries-java/#client
-default: pulsar://localhost:6550
-# default: 10000
+client.serviceUrl: pulsar://localhost:6650
 client.connectionTimeoutMs = 5000
 
 
@@ -31,8 +30,19 @@ client.connectionTimeoutMs = 5000
 # http://pulsar.apache.org/docs/en/client-libraries-java/#configure-producer
 producer.producerName =
 producer.topicName = persistent://public/default/mynbtest
-#producer.sendTimeoutMs =
+producer.sendTimeoutMs =
 
 ### Consumer related configurations (global) - consumer.xxx
 # http://pulsar.apache.org/docs/en/client-libraries-java/#configure-consumer
+consumer.topicNames =
+consumer.topicsPattern =
+consumer.subscriptionName =
+consumer.subscriptionType =
+consumer.consumerName =
+consumer.receiverQueueSize =
 
+### Reader related configurations (global) - reader.xxx
+# https://pulsar.apache.org/docs/en/client-libraries-java/#reader
+reader.topicName =
+reader.receiverQueueSize =
+reader.readerName =
diff --git a/driver-pulsar/src/main/resources/activities/pulsar.yaml b/driver-pulsar/src/main/resources/activities/pulsar.yaml
index 5e338c057..4a7fa026f 100644
--- a/driver-pulsar/src/main/resources/activities/pulsar.yaml
+++ b/driver-pulsar/src/main/resources/activities/pulsar.yaml
@@ -22,7 +22,7 @@ blocks:
 
   - name: producer-block
     tags:
-      type: producer
+      op-type: producer
     statements:
       - producer-stuff:
           #######
@@ -40,14 +40,17 @@ blocks:
                 "ReadingValue": {reading_value}
             }
 
-#  - name: consumer-block
-#    tags:
-#      type: consumer
-#    statements:
-#      - consumer-stuff:
+  - name: consumer-block
+    tags:
+      op-type: consumer
+    statements:
+      - consumer-stuff:
+          topic-names: "persistent://public/default/nbpulsar, persistent://public/default/mynbtest"
+#          topics-pattern: "public/default/.*"
 #          subscription-name:
 #          subscription-type:
-#
+#          consumer-name:
+
 #  - reader:
 #    tags:
 #      type: reader
diff --git a/driver-pulsar/src/main/resources/pulsar.md b/driver-pulsar/src/main/resources/pulsar.md
index bb1078a33..00deabf03 100644
--- a/driver-pulsar/src/main/resources/pulsar.md
+++ b/driver-pulsar/src/main/resources/pulsar.md
@@ -121,7 +121,7 @@ In the above statement block, there are 4 key statement parameters to provide va
           ```
         * At the moment, only "**\<short-topic-name\>**" part can be dynamically bound (e.g. through NB binding function). All other parts must be static values and the corresponding tenants and namespaces must be created in the Pulsar cluster in advance.
 
-**TODO**: allow dynamic binding for "\<tenant-name\>" and "\<namespace-name\>" after adding a phase for creating "\<tenant-name\>" and/or "\<namespace-name\>", similar to C* CQL schema creation phase.!
+**TODO**: allow dynamic binding for "\<tenant-name\>" and "\<namespace-name\>" after adding a phase for creating "\<tenant-name\>" and/or "\<namespace-name\>", similar to C* CQL schema creation phase!
 
 * **msg-key**: Pulsar message key
     * **Optional**
@@ -149,7 +149,7 @@ Pulsar has built-in schema support. Other than primitive types, Pulsar also supp
   schema.definition:
   ```
 
-For the previous Producer block statement example, the **msg-value** parameter has the value of a JSON string that follows the following Avro schema definition (e.g. from a file **iot-example.asvc**)
+For the previous Producer block statement example, the **msg-value** parameter has the value of a JSON string that follows the following Avro schema definition (e.g. as in the sample schema definition file: **[iot-example.asvc](activities/iot-example.avsc)**)
 ```json
 {
   "type": "record",
@@ -166,7 +166,7 @@ For the previous Producer block statement example, the **msg-value** parameter h
 
 ## 1.5. Activity Parameters
 
-At the moment, the following Activity Parameter is supported:
+At the moment, the following Pulsar driver specific Activity Parameter is supported:
 
 - * config=<file/path/to/global/configuration/properties/file>
 
@@ -189,7 +189,7 @@ At the moment, the following Activity Parameter is supported:
 
 To summarize, the original caching design has the following key requirements:
 * **Requirement 1**: Each NB Pulsar activity is able to launch and cache multiple **client spaces**
-* **Requirement 2**:Each client space can launch and cache multiple Pulsar operators of the same type (producer, consumer, etc.)
+* **Requirement 2**: Each client space can launch and cache multiple Pulsar operators of the same type (producer, consumer, etc.)
 
 In the current implementation, only requirement 2 is implemented. Regarding requirement 1, the current implementation only supports one client space per NB Pulsar activity!
 
diff --git a/engine-api/src/main/java/io/nosqlbench/engine/api/activityconfig/yaml/StmtDef.java b/engine-api/src/main/java/io/nosqlbench/engine/api/activityconfig/yaml/StmtDef.java
index 1aea3e5da..aae92d9ed 100644
--- a/engine-api/src/main/java/io/nosqlbench/engine/api/activityconfig/yaml/StmtDef.java
+++ b/engine-api/src/main/java/io/nosqlbench/engine/api/activityconfig/yaml/StmtDef.java
@@ -79,12 +79,14 @@ public class StmtDef implements OpTemplate {
         Map<String, T> map = new LinkedHashMap<>();
         for (String pname : getParams().keySet()) {
             Object object = getParams().get(pname);
-            if (type.isAssignableFrom(object.getClass())) {
-                map.put(pname, type.cast(object));
-            } else {
-                throw new RuntimeException("With param named '" + pname + "" +
+            if (object != null) {
+                if (type.isAssignableFrom(object.getClass())) {
+                    map.put(pname, type.cast(object));
+                } else {
+                    throw new RuntimeException("With param named '" + pname + "" +
                         "' You can't assign an object of type '" + object.getClass().getSimpleName() + "" +
                         "' to '" + type.getSimpleName() + "'. Maybe the YAML format is suggesting the wrong type.");
+                }
             }
         }
         return map;