pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java - pulsar - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */
 package org.apache.pulsar.testclient;

 import static org.apache.commons.lang3.StringUtils.isBlank;
 import static org.apache.commons.lang3.StringUtils.isNotBlank;

 import java.io.FileInputStream;
 import java.text.DecimalFormat;
 import java.util.List;
 import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.LongAdder;

 import org.apache.pulsar.client.api.ClientBuilder;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Reader;
 import org.apache.pulsar.client.api.ReaderBuilder;
 import org.apache.pulsar.client.api.ReaderListener;
 import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.beust.jcommander.JCommander;
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.ParameterException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectWriter;
 import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.RateLimiter;

 public class PerformanceReader {
     private static final LongAdder messagesReceived = new LongAdder();
     private static final LongAdder bytesReceived = new LongAdder();
     private static final DecimalFormat dec = new DecimalFormat("0.000");

     static class Arguments {

         @Parameter(names = { "-h", "--help" }, description = "Help message", help = true)
         boolean help;

         @Parameter(names = { "--conf-file" }, description = "Configuration file")
         public String confFile;

         @Parameter(description = "persistent://prop/ns/my-topic", required = true)
         public List<String> topic;

         @Parameter(names = { "-t", "--num-topics" }, description = "Number of topics")
         public int numTopics = 1;

         @Parameter(names = { "-r", "--rate" }, description = "Simulate a slow message reader (rate in msg/s)")
         public double rate = 0;

         @Parameter(names = { "-m",
                 "--start-message-id" }, description = "Start message id. This can be either 'earliest', 'latest' or a specific message id by using 'lid:eid'")
         public String startMessageId = "earliest";

         @Parameter(names = { "-q", "--receiver-queue-size" }, description = "Size of the receiver queue")
         public int receiverQueueSize = 1000;

         @Parameter(names = { "-c",
                 "--max-connections" }, description = "Max number of TCP connections to a single broker")
         public int maxConnections = 100;

         @Parameter(names = { "-i",
                 "--stats-interval-seconds" }, description = "Statistics Interval Seconds. If 0, statistics will be disabled")
         public long statsIntervalSeconds = 0;

         @Parameter(names = { "-u", "--service-url" }, description = "Pulsar Service URL")
         public String serviceURL;

         @Parameter(names = { "--auth-plugin" }, description = "Authentication plugin class name")
         public String authPluginClassName;

         @Parameter(names = { "--listener-name" }, description = "Listener name for the broker.")
         String listenerName = null;

         @Parameter(
             names = { "--auth-params" },
             description = "Authentication parameters, whose format is determined by the implementation " +
                 "of method `configure` in authentication plugin class, for example \"key1:val1,key2:val2\" " +
                 "or \"{\"key1\":\"val1\",\"key2\":\"val2\"}.")
         public String authParams;

         @Parameter(names = {
                 "--use-tls" }, description = "Use TLS encryption on the connection")
         public boolean useTls;

         @Parameter(names = {
                 "--trust-cert-file" }, description = "Path for the trusted TLS certificate file")
         public String tlsTrustCertsFilePath = "";

         @Parameter(names = {
                 "--tls-allow-insecure" }, description = "Allow insecure TLS connection")
         public Boolean tlsAllowInsecureConnection = null;

         @Parameter(names = { "-time",
                 "--test-duration" }, description = "Test duration in secs. If 0, it will keep consuming")
         public long testTime = 0;

         @Parameter(names = {"-ioThreads", "--num-io-threads"}, description = "Set the number of threads to be " +
                 "used for handling connections to brokers, default is 1 thread")
         public int ioThreads = 1;
     }

     public static void main(String[] args) throws Exception {
         final Arguments arguments = new Arguments();
         JCommander jc = new JCommander(arguments);
         jc.setProgramName("pulsar-perf read");

         try {
             jc.parse(args);
         } catch (ParameterException e) {
             System.out.println(e.getMessage());
             jc.usage();
             System.exit(-1);
         }

         if (arguments.help) {
             jc.usage();
             System.exit(-1);
         }

         if (arguments.topic.size() != 1) {
             System.out.println("Only one topic name is allowed");
             jc.usage();
             System.exit(-1);
         }

         if (arguments.confFile != null) {
             Properties prop = new Properties(System.getProperties());
             prop.load(new FileInputStream(arguments.confFile));

             if (arguments.serviceURL == null) {
                 arguments.serviceURL = prop.getProperty("brokerServiceUrl");
             }

             if (arguments.serviceURL == null) {
                 arguments.serviceURL = prop.getProperty("webServiceUrl");
             }

             // fallback to previous-version serviceUrl property to maintain backward-compatibility
             if (arguments.serviceURL == null) {
                 arguments.serviceURL = prop.getProperty("serviceUrl", "http://localhost:8080/");
             }

             if (arguments.authPluginClassName == null) {
                 arguments.authPluginClassName = prop.getProperty("authPlugin", null);
             }

             if (arguments.authParams == null) {
                 arguments.authParams = prop.getProperty("authParams", null);
             }

             if (arguments.useTls == false) {
                 arguments.useTls = Boolean.parseBoolean(prop.getProperty("useTls"));
             }

             if (isBlank(arguments.tlsTrustCertsFilePath)) {
                 arguments.tlsTrustCertsFilePath = prop.getProperty("tlsTrustCertsFilePath", "");
             }

             if (arguments.tlsAllowInsecureConnection == null) {
                 arguments.tlsAllowInsecureConnection = Boolean.parseBoolean(prop
                         .getProperty("tlsAllowInsecureConnection", ""));
             }
         }

         // Dump config variables
         PerfClientUtils.printJVMInformation(log);
         ObjectMapper m = new ObjectMapper();
         ObjectWriter w = m.writerWithDefaultPrettyPrinter();
         log.info("Starting Pulsar performance reader with config: {}", w.writeValueAsString(arguments));

         final TopicName prefixTopicName = TopicName.get(arguments.topic.get(0));

         final RateLimiter limiter = arguments.rate > 0 ? RateLimiter.create(arguments.rate) : null;
         long startTime = System.nanoTime();
         long testEndTime = startTime + (long) (arguments.testTime * 1e9);
         ReaderListener<byte[]> listener = (reader, msg) -> {
             if (arguments.testTime > 0) {
                 if (System.nanoTime() > testEndTime) {
                     log.info("------------------- DONE -----------------------");
                     System.exit(0);
                 }
             }
             messagesReceived.increment();
             bytesReceived.add(msg.getData().length);

             if (limiter != null) {
                 limiter.acquire();
             }
         };

         ClientBuilder clientBuilder = PulsarClient.builder() //
                 .serviceUrl(arguments.serviceURL) //
                 .connectionsPerBroker(arguments.maxConnections) //
                 .statsInterval(arguments.statsIntervalSeconds, TimeUnit.SECONDS) //
                 .ioThreads(arguments.ioThreads) //
                 .enableTls(arguments.useTls) //
                 .tlsTrustCertsFilePath(arguments.tlsTrustCertsFilePath);

         if (isNotBlank(arguments.authPluginClassName)) {
             clientBuilder.authentication(arguments.authPluginClassName, arguments.authParams);
         }

         if (arguments.tlsAllowInsecureConnection != null) {
             clientBuilder.allowTlsInsecureConnection(arguments.tlsAllowInsecureConnection);
         }

         if (isNotBlank(arguments.listenerName)) {
             clientBuilder.listenerName(arguments.listenerName);
         }

         PulsarClient pulsarClient = clientBuilder.build();

         List<CompletableFuture<Reader<byte[]>>> futures = Lists.newArrayList();

         MessageId startMessageId;
         if ("earliest".equals(arguments.startMessageId)) {
             startMessageId = MessageId.earliest;
         } else if ("latest".equals(arguments.startMessageId)) {
             startMessageId = MessageId.latest;
         } else {
             String[] parts = arguments.startMessageId.split(":");
             startMessageId = new MessageIdImpl(Long.parseLong(parts[0]), Long.parseLong(parts[1]), -1);
         }

         ReaderBuilder<byte[]> readerBuilder = pulsarClient.newReader() //
                 .readerListener(listener) //
                 .receiverQueueSize(arguments.receiverQueueSize) //
                 .startMessageId(startMessageId);

         for (int i = 0; i < arguments.numTopics; i++) {
             final TopicName topicName = (arguments.numTopics == 1) ? prefixTopicName
                     : TopicName.get(String.format("%s-%d", prefixTopicName, i));

             futures.add(readerBuilder.clone().topic(topicName.toString()).createAsync());
         }

         FutureUtil.waitForAll(futures).get();

         log.info("Start reading from {} topics", arguments.numTopics);

         long oldTime = System.nanoTime();

         while (true) {
             try {
                 Thread.sleep(10000);
             } catch (InterruptedException e) {
                 break;
             }

             long now = System.nanoTime();
             double elapsed = (now - oldTime) / 1e9;
             double rate = messagesReceived.sumThenReset() / elapsed;
             double throughput = bytesReceived.sumThenReset() / elapsed * 8 / 1024 / 1024;

             log.info("Read throughput: {}  msg/s -- {} Mbit/s", dec.format(rate), dec.format(throughput));
             oldTime = now;
         }

         pulsarClient.close();
     }

     private static final Logger log = LoggerFactory.getLogger(PerformanceReader.class);
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/
	package org.apache.pulsar.testclient;

	import static org.apache.commons.lang3.StringUtils.isBlank;
	import static org.apache.commons.lang3.StringUtils.isNotBlank;

	import java.io.FileInputStream;
	import java.text.DecimalFormat;
	import java.util.List;
	import java.util.Properties;
	import java.util.concurrent.CompletableFuture;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.atomic.LongAdder;

	import org.apache.pulsar.client.api.ClientBuilder;
	import org.apache.pulsar.client.api.MessageId;
	import org.apache.pulsar.client.api.PulsarClient;
	import org.apache.pulsar.client.api.Reader;
	import org.apache.pulsar.client.api.ReaderBuilder;
	import org.apache.pulsar.client.api.ReaderListener;
	import org.apache.pulsar.client.impl.MessageIdImpl;
	import org.apache.pulsar.common.naming.TopicName;
	import org.apache.pulsar.common.util.FutureUtil;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.beust.jcommander.JCommander;
	import com.beust.jcommander.Parameter;
	import com.beust.jcommander.ParameterException;
	import com.fasterxml.jackson.databind.ObjectMapper;
	import com.fasterxml.jackson.databind.ObjectWriter;
	import com.google.common.collect.Lists;
	import com.google.common.util.concurrent.RateLimiter;

	public class PerformanceReader {
	private static final LongAdder messagesReceived = new LongAdder();
	private static final LongAdder bytesReceived = new LongAdder();
	private static final DecimalFormat dec = new DecimalFormat("0.000");

	static class Arguments {

	@Parameter(names = { "-h", "--help" }, description = "Help message", help = true)
	boolean help;

	@Parameter(names = { "--conf-file" }, description = "Configuration file")
	public String confFile;

	@Parameter(description = "persistent://prop/ns/my-topic", required = true)
	public List<String> topic;

	@Parameter(names = { "-t", "--num-topics" }, description = "Number of topics")
	public int numTopics = 1;

	@Parameter(names = { "-r", "--rate" }, description = "Simulate a slow message reader (rate in msg/s)")
	public double rate = 0;

	@Parameter(names = { "-m",
	"--start-message-id" }, description = "Start message id. This can be either 'earliest', 'latest' or a specific message id by using 'lid:eid'")
	public String startMessageId = "earliest";

	@Parameter(names = { "-q", "--receiver-queue-size" }, description = "Size of the receiver queue")
	public int receiverQueueSize = 1000;

	@Parameter(names = { "-c",
	"--max-connections" }, description = "Max number of TCP connections to a single broker")
	public int maxConnections = 100;

	@Parameter(names = { "-i",
	"--stats-interval-seconds" }, description = "Statistics Interval Seconds. If 0, statistics will be disabled")
	public long statsIntervalSeconds = 0;

	@Parameter(names = { "-u", "--service-url" }, description = "Pulsar Service URL")
	public String serviceURL;

	@Parameter(names = { "--auth-plugin" }, description = "Authentication plugin class name")
	public String authPluginClassName;

	@Parameter(names = { "--listener-name" }, description = "Listener name for the broker.")
	String listenerName = null;

	@Parameter(
	names = { "--auth-params" },
	description = "Authentication parameters, whose format is determined by the implementation " +
	"of method `configure` in authentication plugin class, for example \"key1:val1,key2:val2\" " +
	"or \"{\"key1\":\"val1\",\"key2\":\"val2\"}.")
	public String authParams;

	@Parameter(names = {
	"--use-tls" }, description = "Use TLS encryption on the connection")
	public boolean useTls;

	@Parameter(names = {
	"--trust-cert-file" }, description = "Path for the trusted TLS certificate file")
	public String tlsTrustCertsFilePath = "";

	@Parameter(names = {
	"--tls-allow-insecure" }, description = "Allow insecure TLS connection")
	public Boolean tlsAllowInsecureConnection = null;

	@Parameter(names = { "-time",
	"--test-duration" }, description = "Test duration in secs. If 0, it will keep consuming")
	public long testTime = 0;

	@Parameter(names = {"-ioThreads", "--num-io-threads"}, description = "Set the number of threads to be " +
	"used for handling connections to brokers, default is 1 thread")
	public int ioThreads = 1;
	}

	public static void main(String[] args) throws Exception {
	final Arguments arguments = new Arguments();
	JCommander jc = new JCommander(arguments);
	jc.setProgramName("pulsar-perf read");

	try {
	jc.parse(args);
	} catch (ParameterException e) {
	System.out.println(e.getMessage());
	jc.usage();
	System.exit(-1);
	}

	if (arguments.help) {
	jc.usage();
	System.exit(-1);
	}

	if (arguments.topic.size() != 1) {
	System.out.println("Only one topic name is allowed");
	jc.usage();
	System.exit(-1);
	}

	if (arguments.confFile != null) {
	Properties prop = new Properties(System.getProperties());
	prop.load(new FileInputStream(arguments.confFile));

	if (arguments.serviceURL == null) {
	arguments.serviceURL = prop.getProperty("brokerServiceUrl");
	}

	if (arguments.serviceURL == null) {
	arguments.serviceURL = prop.getProperty("webServiceUrl");
	}

	// fallback to previous-version serviceUrl property to maintain backward-compatibility
	if (arguments.serviceURL == null) {
	arguments.serviceURL = prop.getProperty("serviceUrl", "http://localhost:8080/");
	}

	if (arguments.authPluginClassName == null) {
	arguments.authPluginClassName = prop.getProperty("authPlugin", null);
	}

	if (arguments.authParams == null) {
	arguments.authParams = prop.getProperty("authParams", null);
	}

	if (arguments.useTls == false) {
	arguments.useTls = Boolean.parseBoolean(prop.getProperty("useTls"));
	}

	if (isBlank(arguments.tlsTrustCertsFilePath)) {
	arguments.tlsTrustCertsFilePath = prop.getProperty("tlsTrustCertsFilePath", "");
	}

	if (arguments.tlsAllowInsecureConnection == null) {
	arguments.tlsAllowInsecureConnection = Boolean.parseBoolean(prop
	.getProperty("tlsAllowInsecureConnection", ""));
	}
	}

	// Dump config variables
	PerfClientUtils.printJVMInformation(log);
	ObjectMapper m = new ObjectMapper();
	ObjectWriter w = m.writerWithDefaultPrettyPrinter();
	log.info("Starting Pulsar performance reader with config: {}", w.writeValueAsString(arguments));

	final TopicName prefixTopicName = TopicName.get(arguments.topic.get(0));

	final RateLimiter limiter = arguments.rate > 0 ? RateLimiter.create(arguments.rate) : null;
	long startTime = System.nanoTime();
	long testEndTime = startTime + (long) (arguments.testTime * 1e9);
	ReaderListener<byte[]> listener = (reader, msg) -> {
	if (arguments.testTime > 0) {
	if (System.nanoTime() > testEndTime) {
	log.info("------------------- DONE -----------------------");
	System.exit(0);
	}
	}
	messagesReceived.increment();
	bytesReceived.add(msg.getData().length);

	if (limiter != null) {
	limiter.acquire();
	}
	};

	ClientBuilder clientBuilder = PulsarClient.builder() //
	.serviceUrl(arguments.serviceURL) //
	.connectionsPerBroker(arguments.maxConnections) //
	.statsInterval(arguments.statsIntervalSeconds, TimeUnit.SECONDS) //
	.ioThreads(arguments.ioThreads) //
	.enableTls(arguments.useTls) //
	.tlsTrustCertsFilePath(arguments.tlsTrustCertsFilePath);

	if (isNotBlank(arguments.authPluginClassName)) {
	clientBuilder.authentication(arguments.authPluginClassName, arguments.authParams);
	}

	if (arguments.tlsAllowInsecureConnection != null) {
	clientBuilder.allowTlsInsecureConnection(arguments.tlsAllowInsecureConnection);
	}

	if (isNotBlank(arguments.listenerName)) {
	clientBuilder.listenerName(arguments.listenerName);
	}

	PulsarClient pulsarClient = clientBuilder.build();

	List<CompletableFuture<Reader<byte[]>>> futures = Lists.newArrayList();

	MessageId startMessageId;
	if ("earliest".equals(arguments.startMessageId)) {
	startMessageId = MessageId.earliest;
	} else if ("latest".equals(arguments.startMessageId)) {
	startMessageId = MessageId.latest;
	} else {
	String[] parts = arguments.startMessageId.split(":");
	startMessageId = new MessageIdImpl(Long.parseLong(parts[0]), Long.parseLong(parts[1]), -1);
	}

	ReaderBuilder<byte[]> readerBuilder = pulsarClient.newReader() //
	.readerListener(listener) //
	.receiverQueueSize(arguments.receiverQueueSize) //
	.startMessageId(startMessageId);

	for (int i = 0; i < arguments.numTopics; i++) {
	final TopicName topicName = (arguments.numTopics == 1) ? prefixTopicName
	: TopicName.get(String.format("%s-%d", prefixTopicName, i));

	futures.add(readerBuilder.clone().topic(topicName.toString()).createAsync());
	}

	FutureUtil.waitForAll(futures).get();

	log.info("Start reading from {} topics", arguments.numTopics);

	long oldTime = System.nanoTime();

	while (true) {
	try {
	Thread.sleep(10000);
	} catch (InterruptedException e) {
	break;
	}

	long now = System.nanoTime();
	double elapsed = (now - oldTime) / 1e9;
	double rate = messagesReceived.sumThenReset() / elapsed;
	double throughput = bytesReceived.sumThenReset() / elapsed * 8 / 1024 / 1024;

	log.info("Read throughput: {} msg/s -- {} Mbit/s", dec.format(rate), dec.format(throughput));
	oldTime = now;
	}

	pulsarClient.close();
	}

	private static final Logger log = LoggerFactory.getLogger(PerformanceReader.class);
	}