[TEST-6] Read after write latency testing.

benchmark-framework/pom.xml

@@ -192,6 +192,18 @@
 			<version>4.1.50.Final</version>
 		</dependency>
 
+		<dependency>
+			<groupId>io.netty</groupId>
+		<artifactId>netty-common</artifactId>
+		<version>4.1.50.Final</version>
+	</dependency>
+
+	<dependency>
+		<groupId>io.netty</groupId>
+		<artifactId>netty-buffer</artifactId>
+		<version>4.1.50.Final</version>
+	</dependency>
+
 		<dependency>
 			<groupId>junit</groupId>
 			<artifactId>junit</artifactId>

benchmark-framework/src/main/java/io/openmessaging/benchmark/Benchmark.java

@@ -67,6 +67,7 @@ static class Arguments {
 
     public static void main(String[] args) throws Exception {
         final Arguments arguments = new Arguments();
+        System.out.println("Arguments" + arguments.toString());
         JCommander jc = new JCommander(arguments);
         jc.setProgramName("messaging-benchmark");
 

benchmark-framework/src/main/java/io/openmessaging/benchmark/WorkloadGenerator.java

@@ -142,19 +142,19 @@ private void ensureTopicsAreReady() throws IOException {
         // In this case we just publish 1 message and then wait for consumers to receive the data
         worker.probeProducers();
 
-        while (true) {
-            CountersStats stats = worker.getCountersStats();
-
-            if (stats.messagesReceived < expectedMessages) {
-                try {
-                    Thread.sleep(100);
-                } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                }
-            } else {
-                break;
-            }
-        }
+//        while (true) {
+//            CountersStats stats = worker.getCountersStats();
+//
+//            if (stats.messagesReceived < expectedMessages) {
+//                try {
+//                    Thread.sleep(100);
+//                } catch (InterruptedException e) {
+//                    throw new RuntimeException(e);
+//                }
+//            } else {
+//                break;
+//            }
+//        }
 
         log.info("All consumers are ready");
     }

benchmark-framework/src/main/resources/log4j2.yaml

@@ -34,5 +34,5 @@ Configuration:
       additivity: false
       AppenderRef:
         ref: Console
-        
+
 

driver-pravega/src/main/java/io/openmessaging/benchmark/driver/pravega/PravegaBenchmarkConsumer.java

@@ -32,7 +32,9 @@
 import org.slf4j.LoggerFactory;
 
 import java.nio.ByteBuffer;
-import java.util.UUID;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -44,10 +46,13 @@ public class PravegaBenchmarkConsumer implements BenchmarkConsumer {
     private final ExecutorService executor;
     private final EventStreamReader<ByteBuffer> reader;
     private final AtomicBoolean closed = new AtomicBoolean(false);
+    private ConcurrentHashMap<String, List<Long>> txnsWithEvents;
+
 
     public PravegaBenchmarkConsumer(String streamName, String scopeName, String subscriptionName, ConsumerCallback consumerCallback,
                                     EventStreamClientFactory clientFactory, ReaderGroupManager readerGroupManager,
-                                    boolean includeTimestampInEvent) {
+                                    boolean includeTimestampInEvent, ConcurrentHashMap txnsWithEvents) {
+        this.txnsWithEvents = txnsWithEvents;
         log.info("PravegaBenchmarkConsumer: BEGIN: subscriptionName={}, streamName={}", subscriptionName, streamName);
         // Create reader group if it doesn't already exist.
         final ReaderGroupConfig readerGroupConfig = ReaderGroupConfig.builder()
@@ -66,16 +71,60 @@ public PravegaBenchmarkConsumer(String streamName, String scopeName, String subs
            while (!closed.get()) {
                try {
                    final ByteBuffer event = reader.readNextEvent(1000).getEvent();
+                   final long eventInTxnReadTs = System.nanoTime();
                    if (event != null) {
-                       long eventTimestamp;
-                       if (includeTimestampInEvent) {
-                           eventTimestamp = event.getLong();
-                       } else {
-                           // This will result in an invalid end-to-end latency measurement of 0 seconds.
-                           eventTimestamp = TimeUnit.MICROSECONDS.toMillis(Long.MAX_VALUE);
+                       long eventTimestamp = TimeUnit.MICROSECONDS.toMillis(Long.MAX_VALUE);;
+//                       if (includeTimestampInEvent) {
+//                           eventTimestamp = event.getLong();
+//                       } else {
+//                           // This will result in an invalid end-to-end latency measurement of 0 seconds.
+//                           eventTimestamp = TimeUnit.MICROSECONDS.toMillis(Long.MAX_VALUE);
+//                       }
+                       final String test = new String(event.array(), event.arrayOffset() + event.position(), event.remaining(), StandardCharsets.UTF_16);
+                       final String[] payloadComponents = test.split("---");
+                       if (payloadComponents.length < 2) {
+                           continue;
                        }
+                       final String txnId = payloadComponents[0];
+                       final long txnWriteTs = Long.parseLong(payloadComponents[1]);
+                       final long readDurationMs = (eventInTxnReadTs - txnWriteTs) / (long) 1000000;
+                       this.txnsWithEvents.compute(txnId, (k,v) -> {
+                           if (v == null) {
+                               v = new LinkedList<Long>();
+                           }
+                           v.add(readDurationMs);
+                           if (v.size() == 1000) {
+                               double summary = 0.0;
+                               for (Iterator<Long> i = v.iterator(); i.hasNext();) {
+                                   long currentTime = i.next();
+                                   summary += currentTime;
+                               }
+                               double average = (summary) / v.size();
+                               log.info("READTXN---ID---" + txnId + "---AVG---" + average);
+
+                           }
+                           return v;
+                       });
+//                       List<Long> readEventsInTxn = this.txnsWithEvents.putIfAbsent(txnId, Arrays.asList(readDurationMs));
+//                       if (readEventsInTxn != null) {
+//                           readEventsInTxn.add(txnWriteTs);
+//                           // TODO: make the value configurable
+//                           if (readEventsInTxn.size() == 1000) {
+//                               double summary = 0.0;
+//                               for (Iterator<Long> i = readEventsInTxn.iterator(); i.hasNext();) {
+//                                   long currentTime = i.next();
+//                                   summary += currentTime;
+//                               }
+//                               double average = (summary) / readEventsInTxn.size();
+//                               log.info("READTXN---ID---" + txnId + "---AVG---" + average);
+//
+//                           }
+//                       }
+//                       log.info("Transaction " + payloadComponents[0] + "---" + readDurationMs);
                        byte[] payload = new byte[event.remaining()];
-                       event.get(payload);
+//                       String payloadStr = new String(payload, StandardCharsets.UTF_16);
+//                       event.get(payload);
+
                        consumerCallback.messageReceived(payload, eventTimestamp);
                    }
                } catch (ReinitializationRequiredException e) {

driver-pravega/src/main/java/io/openmessaging/benchmark/driver/pravega/PravegaBenchmarkDriver.java

@@ -43,6 +43,8 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class PravegaBenchmarkDriver implements BenchmarkDriver {
     private static final Logger log = LoggerFactory.getLogger(PravegaBenchmarkDriver.class);
@@ -56,17 +58,23 @@ public class PravegaBenchmarkDriver implements BenchmarkDriver {
     private ReaderGroupManager readerGroupManager;
     private EventStreamClientFactory clientFactory;
     private final List<String> createdTopics = new ArrayList<>();
+    private ConcurrentHashMap<String, List> txnsWithEvents;
+    private AtomicInteger totalTxnAmount;
+
 
     @Override
     public void initialize(File configurationFile, StatsLogger statsLogger) throws IOException {
         config = readConfig(configurationFile);
         log.info("Pravega driver configuration: {}", objectWriter.writeValueAsString(config));
-
-        clientConfig = ClientConfig.builder().controllerURI(URI.create(config.client.controllerURI)).build();
+        clientConfig = ClientConfig.builder().controllerURI(URI.create(config.client.controllerURI)).validateHostName(false).build();
         scopeName = config.client.scopeName;
         streamManager = StreamManager.create(clientConfig);
         readerGroupManager = ReaderGroupManager.withScope(scopeName, clientConfig);
         clientFactory = EventStreamClientFactory.withScope(scopeName, clientConfig);
+
+        // Initialize fields for read-after-write measurement purposes
+        this.txnsWithEvents = new ConcurrentHashMap<>();
+        this.totalTxnAmount = new AtomicInteger();
     }
 
     private static final ObjectMapper mapper = new ObjectMapper(new YAMLFactory())
@@ -118,7 +126,7 @@ public CompletableFuture<BenchmarkProducer> createProducer(String topic) {
         BenchmarkProducer producer = null;
         if (config.enableTransaction) {
             producer = new PravegaBenchmarkTransactionProducer(topic, clientFactory, config.includeTimestampInEvent,
-                    config.writer.enableConnectionPooling, config.eventsPerTransaction);
+                    config.writer.enableConnectionPooling, config.eventsPerTransaction, this.totalTxnAmount);
         } else {
             producer = new PravegaBenchmarkProducer(topic, clientFactory, config.includeTimestampInEvent,
                     config.writer.enableConnectionPooling);
@@ -132,11 +140,12 @@ public CompletableFuture<BenchmarkConsumer> createConsumer(String topic, String
         topic = cleanName(topic);
         subscriptionName = cleanName(subscriptionName);
         BenchmarkConsumer consumer = new PravegaBenchmarkConsumer(topic, scopeName, subscriptionName, consumerCallback,
-                clientFactory, readerGroupManager, config.includeTimestampInEvent);
+                clientFactory, readerGroupManager, config.includeTimestampInEvent, this.txnsWithEvents);
         return CompletableFuture.completedFuture(consumer);
     }
 
     private void deleteTopics() {
+        log.info("---CREATED-AMM-OF-TXN-FROM-ALL-PRODUCERS---" + this.totalTxnAmount.get());
         synchronized (createdTopics) {
             for (String topic : createdTopics) {
                 log.info("deleteTopics: topic={}", topic);

driver-pravega/src/main/java/io/openmessaging/benchmark/driver/pravega/PravegaBenchmarkProducer.java

@@ -46,6 +46,7 @@ public PravegaBenchmarkProducer(String streamName, EventStreamClientFactory clie
                 new ByteBufferSerializer(),
                 EventWriterConfig.builder()
                         .enableConnectionPooling(enableConnectionPooling)
+                        .transactionTimeoutTime(120000)
                         .build());
         this.includeTimestampInEvent = includeTimestampInEvent;
     }