Pulsar: add bytes counter

driver-pulsar/pom.xml

@@ -27,7 +27,7 @@
 
         <dependency>
             <groupId>org.apache.pulsar</groupId>
-            <artifactId>pulsar-client</artifactId>
+            <artifactId>pulsar-client-original</artifactId>
             <version>${pulsar.version}</version>
         </dependency>
 
@@ -75,7 +75,7 @@
         <dependency>
             <groupId>org.apache.avro</groupId>
             <artifactId>avro</artifactId>
-            <version>1.10.1</version>
+            <version>1.9.1</version>
         </dependency>
 
     </dependencies>

driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarActivity.java

@@ -1,5 +1,7 @@
 package io.nosqlbench.driver.pulsar;
 
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Meter;
 import com.codahale.metrics.Timer;
 import io.nosqlbench.driver.pulsar.ops.PulsarOp;
 import io.nosqlbench.driver.pulsar.ops.ReadyPulsarOp;
@@ -20,6 +22,7 @@ public class PulsarActivity extends SimpleActivity implements ActivityDefObserve
 
     public Timer bindTimer;
     public Timer executeTimer;
+    public Counter bytesCounter;
     private PulsarSpaceCache pulsarCache;
 
     private PulsarNBClientConf clientConf;
@@ -41,6 +44,7 @@ public class PulsarActivity extends SimpleActivity implements ActivityDefObserve
 
         bindTimer = ActivityMetrics.timer(activityDef, "bind");
         executeTimer = ActivityMetrics.timer(activityDef, "execute");
+        bytesCounter = ActivityMetrics.counter(activityDef, "bytes");
 
         String pulsarClntConfFile = activityDef.getParams().getOptionalString("config").orElse("config.properties");
         clientConf = new PulsarNBClientConf(pulsarClntConfFile);
@@ -49,7 +53,7 @@ public class PulsarActivity extends SimpleActivity implements ActivityDefObserve
 
         pulsarCache = new PulsarSpaceCache(this);
 
-        this.sequencer = createOpSequence((ot) -> new ReadyPulsarOp(ot, pulsarCache));
+        this.sequencer = createOpSequence((ot) -> new ReadyPulsarOp(ot, pulsarCache, this));
         setDefaultsFromOpSequence(sequencer);
         onActivityDefUpdate(activityDef);
 
@@ -87,4 +91,8 @@ public class PulsarActivity extends SimpleActivity implements ActivityDefObserve
     public Timer getExecuteTimer() {
         return this.executeTimer;
     }
+
+    public Counter getBytesCounter() {
+        return bytesCounter;
+    }
 }

driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarProducerMapper.java

@@ -1,5 +1,6 @@
 package io.nosqlbench.driver.pulsar.ops;
 
+import com.codahale.metrics.Counter;
 import io.nosqlbench.driver.pulsar.PulsarSpace;
 import io.nosqlbench.engine.api.templating.CommandTemplate;
 import org.apache.pulsar.client.api.Producer;
@@ -22,18 +23,21 @@ public class PulsarProducerMapper extends PulsarOpMapper {
     private final LongFunction<Boolean> asyncApiFunc;
     private final LongFunction<String> keyFunc;
     private final LongFunction<String> payloadFunc;
+    private final Counter bytesCounter;
 
     public PulsarProducerMapper(CommandTemplate cmdTpl,
                                 PulsarSpace clientSpace,
                                 LongFunction<Producer<?>> producerFunc,
                                 LongFunction<Boolean> asyncApiFunc,
                                 LongFunction<String> keyFunc,
-                                LongFunction<String> payloadFunc) {
+                                LongFunction<String> payloadFunc,
+                                Counter bytesCounter) {
         super(cmdTpl, clientSpace);
         this.producerFunc = producerFunc;
         this.asyncApiFunc = asyncApiFunc;
         this.keyFunc = keyFunc;
         this.payloadFunc = payloadFunc;
+        this.bytesCounter = bytesCounter;
     }
 
     @Override
@@ -48,6 +52,7 @@ public class PulsarProducerMapper extends PulsarOpMapper {
             clientSpace.getPulsarSchema(),
             asyncApi,
             msgKey,
-            msgPayload);
+            msgPayload,
+            bytesCounter);
     }
 }

driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarProducerOp.java

@@ -1,6 +1,6 @@
 package io.nosqlbench.driver.pulsar.ops;
 
-import io.nosqlbench.driver.pulsar.PulsarAction;
+import com.codahale.metrics.Counter;
 import io.nosqlbench.driver.pulsar.util.AvroUtil;
 import io.nosqlbench.driver.pulsar.util.PulsarActivityUtil;
 import org.apache.logging.log4j.LogManager;
@@ -22,17 +22,20 @@ public class PulsarProducerOp implements PulsarOp {
     private final String msgKey;
     private final String msgPayload;
     private final boolean asyncPulsarOp;
+    private final Counter bytesCounter;
 
     public PulsarProducerOp(Producer<?> producer,
                             Schema<?> schema,
                             boolean asyncPulsarOp,
                             String key,
-                            String payload) {
+                            String payload,
+                            Counter bytesCounter) {
         this.producer = producer;
         this.pulsarSchema = schema;
         this.msgKey = key;
         this.msgPayload = payload;
         this.asyncPulsarOp = asyncPulsarOp;
+        this.bytesCounter = bytesCounter;
     }
 
     @Override
@@ -54,8 +57,12 @@ public class PulsarProducerOp implements PulsarOp {
                 msgPayload
             );
             typedMessageBuilder = typedMessageBuilder.value(payload);
+            // TODO: add a way to calculate the message size for AVRO messages
+            bytesCounter.inc(msgPayload.length());
         } else {
-            typedMessageBuilder = typedMessageBuilder.value(msgPayload.getBytes(StandardCharsets.UTF_8));
+            byte[] array = msgPayload.getBytes(StandardCharsets.UTF_8);
+            typedMessageBuilder = typedMessageBuilder.value(array);
+            bytesCounter.inc(array.length);
         }
 
         //TODO: add error handling with failed message production

driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/ReadyPulsarOp.java

@@ -20,11 +20,13 @@ public class ReadyPulsarOp implements OpDispenser<PulsarOp> {
     private final CommandTemplate cmdTpl;
     private final PulsarSpace clientSpace;
     private final LongFunction<PulsarOp> opFunc;
+    private final PulsarActivity pulsarActivity;
 
     // TODO: Add docs for the command template with respect to the OpTemplate
 
-    public ReadyPulsarOp(OpTemplate opTemplate, PulsarSpaceCache pcache) {
+    public ReadyPulsarOp(OpTemplate opTemplate, PulsarSpaceCache pcache, PulsarActivity pulsarActivity) {
         // TODO: Consider parsing map structures into equivalent binding representation
+        this.pulsarActivity = pulsarActivity;
         this.opTpl = opTemplate;
         this.cmdTpl = new CommandTemplate(opTemplate);
 
@@ -180,7 +182,8 @@ public class ReadyPulsarOp implements OpDispenser<PulsarOp> {
             producerFunc,
             async_api_func,
             keyFunc,
-            valueFunc);
+            valueFunc,
+            pulsarActivity.getBytesCounter());
     }
 
     private LongFunction<PulsarOp> resolveMsgConsume(