From cd4fda6829e03109a24bd5cb92099927df9fc83f Mon Sep 17 00:00:00 2001 From: Ben Stopford Date: Mon, 7 Sep 2015 17:36:51 -0700 Subject: [PATCH 1/3] KAFKA-2431: Ease testing of SSL - Avoid System.exit in ProducerPerformance so this can be externally invoked - Add command line option so that a default set of properties can be speicfied in ConsumerPerformance (needed for ssl properties) - Allow cipher suites to be specified via properties - Increase timeout in ConsumerPerformance (useful when running with SSL under load) - Use a relative default value for the reporting interval - Add option for show-all-stats (includes intermediary and summary stats) --- .../kafka/common/config/SSLConfigs.java | 2 +- .../kafka/common/security/ssl/SSLFactory.java | 5 +-- .../main/scala/kafka/server/KafkaConfig.scala | 4 +++ .../kafka/tools/ConsumerPerformance.scala | 35 ++++++++++++------- .../main/scala/kafka/tools/PerfConfig.scala | 4 ++- .../kafka/tools/ProducerPerformance.scala | 7 +++- .../unit/kafka/server/KafkaConfigTest.scala | 1 + 7 files changed, 40 insertions(+), 18 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/config/SSLConfigs.java b/clients/src/main/java/org/apache/kafka/common/config/SSLConfigs.java index dd7b71a46c568..2bd34fa161ce8 100644 --- a/clients/src/main/java/org/apache/kafka/common/config/SSLConfigs.java +++ b/clients/src/main/java/org/apache/kafka/common/config/SSLConfigs.java @@ -34,7 +34,7 @@ public class SSLConfigs { public static final String SSL_PROVIDER_DOC = "The name of the security provider used for SSL connections. Default value is the default security provider of the JVM."; public static final String SSL_CIPHER_SUITES_CONFIG = "ssl.cipher.suites"; - public static final String SSL_CIPHER_SUITES_DOC = "A cipher suite is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol." + public static final String SSL_CIPHER_SUITES_DOC = "A list of cipher suites. This is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol." + "By default all the available cipher suites are supported."; public static final String SSL_ENABLED_PROTOCOLS_CONFIG = "ssl.enabled.protocols"; diff --git a/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java b/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java index f79b65cdfd014..db64ab32af4ab 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java +++ b/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java @@ -60,12 +60,13 @@ public void configure(Map configs) throws KafkaException { if (configs.get(SSLConfigs.SSL_CIPHER_SUITES_CONFIG) != null) { List cipherSuitesList = (List) configs.get(SSLConfigs.SSL_CIPHER_SUITES_CONFIG); - this.cipherSuites = (String[]) cipherSuitesList.toArray(new String[cipherSuitesList.size()]); + if (!cipherSuitesList.isEmpty()) + this.cipherSuites = cipherSuitesList.toArray(new String[cipherSuitesList.size()]); } if (configs.get(SSLConfigs.SSL_ENABLED_PROTOCOLS_CONFIG) != null) { List enabledProtocolsList = (List) configs.get(SSLConfigs.SSL_ENABLED_PROTOCOLS_CONFIG); - this.enabledProtocols = (String[]) enabledProtocolsList.toArray(new String[enabledProtocolsList.size()]); + this.enabledProtocols = enabledProtocolsList.toArray(new String[enabledProtocolsList.size()]); } if (configs.containsKey(SSLConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG)) { diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index 46f4a25073855..6b605fc7a6a74 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -176,6 +176,7 @@ object Defaults { val SSLClientAuthRequested = "requested" val SSLClientAuthNone = "none" val SSLClientAuth = SSLClientAuthNone + val SSLCipherSuites = "" } @@ -650,6 +651,7 @@ object KafkaConfig { .define(SSLKeyManagerAlgorithmProp, STRING, Defaults.SSLKeyManagerAlgorithm, MEDIUM, SSLKeyManagerAlgorithmDoc) .define(SSLTrustManagerAlgorithmProp, STRING, Defaults.SSLTrustManagerAlgorithm, MEDIUM, SSLTrustManagerAlgorithmDoc) .define(SSLClientAuthProp, STRING, Defaults.SSLClientAuth, in(Defaults.SSLClientAuthRequired, Defaults.SSLClientAuthRequested, Defaults.SSLClientAuthNone), MEDIUM, SSLClientAuthDoc) + .define(SSLCipherSuitesProp, LIST, Defaults.SSLCipherSuites, MEDIUM, SSLCipherSuitesDoc) } @@ -809,6 +811,7 @@ case class KafkaConfig (props: java.util.Map[_, _]) extends AbstractConfig(Kafka val sslKeyManagerAlgorithm = getString(KafkaConfig.SSLKeyManagerAlgorithmProp) val sslTrustManagerAlgorithm = getString(KafkaConfig.SSLTrustManagerAlgorithmProp) val sslClientAuth = getString(KafkaConfig.SSLClientAuthProp) + val sslCipher = getList(KafkaConfig.SSLCipherSuitesProp) /** ********* Quota Configuration **************/ val producerQuotaBytesPerSecondDefault = getLong(KafkaConfig.ProducerQuotaBytesPerSecondDefaultProp) @@ -948,6 +951,7 @@ case class KafkaConfig (props: java.util.Map[_, _]) extends AbstractConfig(Kafka channelConfigs.put(SSLKeyManagerAlgorithmProp, sslKeyManagerAlgorithm) channelConfigs.put(SSLTrustManagerAlgorithmProp, sslTrustManagerAlgorithm) channelConfigs.put(SSLClientAuthProp, sslClientAuth) + channelConfigs.put(SSLCipherSuitesProp, sslCipher) channelConfigs } diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index 826703072b921..907f37839d35f 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -20,6 +20,7 @@ package kafka.tools import java.util import org.apache.kafka.common.TopicPartition +import org.apache.kafka.common.utils.Utils import scala.collection.JavaConversions._ import java.util.concurrent.atomic.AtomicLong @@ -50,19 +51,15 @@ object ConsumerPerformance { val totalBytesRead = new AtomicLong(0) val consumerTimeout = new AtomicBoolean(false) - if (!config.hideHeader) { - if (!config.showDetailedStats) - println("start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") - else + if (!config.hideHeader && (config.showDetailedStats || config.showAllStats)) println("time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") - } var startMs, endMs = 0L if(config.useNewConsumer) { val consumer = new KafkaConsumer[Array[Byte], Array[Byte]](config.props) consumer.subscribe(List(config.topic)) startMs = System.currentTimeMillis - consume(consumer, List(config.topic), config.numMessages, 1000, config, totalMessagesRead, totalBytesRead) + consume(consumer, List(config.topic), config.numMessages, 5000, config, totalMessagesRead, totalBytesRead) endMs = System.currentTimeMillis consumer.close() } else { @@ -90,8 +87,12 @@ object ConsumerPerformance { consumerConnector.shutdown() } val elapsedSecs = (endMs - startMs) / 1000.0 - if (!config.showDetailedStats) { - val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) + val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) + + if(!config.showDetailedStats || config.showAllStats){ + if (!config.hideHeader ) + println("start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") + println(("%s, %s, %.4f, %.4f, %d, %.4f").format(config.dateFormat.format(startMs), config.dateFormat.format(endMs), totalMBRead, totalMBRead / elapsedSecs, totalMessagesRead.get, totalMessagesRead.get / elapsedSecs)) } @@ -139,7 +140,7 @@ object ConsumerPerformance { bytesRead += record.value.size if (messagesRead % config.reportingInterval == 0) { - if (config.showDetailedStats) + if (config.showDetailedStats || config.showAllStats) printProgressMessage(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, lastReportTime, System.currentTimeMillis, config.dateFormat) lastReportTime = System.currentTimeMillis lastMessagesRead = messagesRead @@ -201,7 +202,12 @@ object ConsumerPerformance { .withRequiredArg .describedAs("count") .ofType(classOf[java.lang.Integer]) - .defaultsTo(1) + .defaultsTo(1) + val propsFileOpt = parser.accepts("consumer.config", "External consumer properties file.") + .withRequiredArg + .describedAs("consumer properties") + .ofType(classOf[java.lang.String]) + .defaultsTo("") val useNewConsumerOpt = parser.accepts("new-consumer", "Use the new consumer implementation.") val options = parser.parse(args: _*) @@ -210,7 +216,7 @@ object ConsumerPerformance { val useNewConsumer = options.has(useNewConsumerOpt) - val props = new Properties + val props = if (options.has(propsFileOpt)) Utils.loadProps(options.valueOf(propsFileOpt)) else new Properties if(useNewConsumer) { import org.apache.kafka.clients.consumer.ConsumerConfig props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServersOpt)) @@ -234,10 +240,13 @@ object ConsumerPerformance { val numThreads = options.valueOf(numThreadsOpt).intValue val topic = options.valueOf(topicOpt) val numMessages = options.valueOf(numMessagesOpt).longValue - val reportingInterval = options.valueOf(reportingIntervalOpt).intValue val showDetailedStats = options.has(showDetailedStatsOpt) + val showAllStats = options.has(showAllStatsOpt) val dateFormat = new SimpleDateFormat(options.valueOf(dateFormatOpt)) val hideHeader = options.has(hideHeaderOpt) + var reportingInterval = options.valueOf(reportingIntervalOpt).longValue + if(reportingInterval == -1) + reportingInterval = numMessages/10 } class ConsumerPerfThread(threadId: Int, name: String, stream: KafkaStream[Array[Byte], Array[Byte]], @@ -260,7 +269,7 @@ object ConsumerPerformance { bytesRead += messageAndMetadata.message.length if (messagesRead % config.reportingInterval == 0) { - if (config.showDetailedStats) + if (config.showDetailedStats || config.showAllStats) printProgressMessage(threadId, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, lastReportTime, System.currentTimeMillis, config.dateFormat) lastReportTime = System.currentTimeMillis lastMessagesRead = messagesRead diff --git a/core/src/main/scala/kafka/tools/PerfConfig.scala b/core/src/main/scala/kafka/tools/PerfConfig.scala index 298bb29fe797f..933d0f7a075ca 100644 --- a/core/src/main/scala/kafka/tools/PerfConfig.scala +++ b/core/src/main/scala/kafka/tools/PerfConfig.scala @@ -30,7 +30,7 @@ class PerfConfig(args: Array[String]) { .withRequiredArg .describedAs("size") .ofType(classOf[java.lang.Integer]) - .defaultsTo(5000) + .defaultsTo(-1) val dateFormatOpt = parser.accepts("date-format", "The date format to use for formatting the time field. " + "See java.text.SimpleDateFormat for options.") .withRequiredArg @@ -39,6 +39,8 @@ class PerfConfig(args: Array[String]) { .defaultsTo("yyyy-MM-dd HH:mm:ss:SSS") val showDetailedStatsOpt = parser.accepts("show-detailed-stats", "If set, stats are reported for each reporting " + "interval as configured by reporting-interval") + val showAllStatsOpt = parser.accepts("show-all-stats", "Show intermediary stats based on the " + + "reporting interval and include the summary stats at the end of the run") val hideHeaderOpt = parser.accepts("hide-header", "If set, skips printing the header for the stats ") val messageSizeOpt = parser.accepts("message-size", "The size of each message.") .withRequiredArg diff --git a/core/src/main/scala/kafka/tools/ProducerPerformance.scala b/core/src/main/scala/kafka/tools/ProducerPerformance.scala index e299f8b26ecce..5f718bcbc9859 100644 --- a/core/src/main/scala/kafka/tools/ProducerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ProducerPerformance.scala @@ -38,6 +38,12 @@ import org.apache.log4j.Logger object ProducerPerformance extends Logging { def main(args: Array[String]) { + run(args) + System.exit(0) + } + + def run(args: Array[String]) { + println("Running ProducerPerformance with args: " + args.map(s => " " + s).deep.mkString) val logger = Logger.getLogger(getClass) val config = new ProducerPerfConfig(args) if (!config.isFixedSize) @@ -66,7 +72,6 @@ object ProducerPerformance extends Logging { config.dateFormat.format(startMs), config.dateFormat.format(endMs), config.compressionCodec.codec, config.messageSize, config.batchSize, totalMBSent, totalMBSent / elapsedSecs, totalMessagesSent.get, totalMessagesSent.get / elapsedSecs)) - System.exit(0) } class ProducerPerfConfig(args: Array[String]) extends PerfConfig(args) { diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala index bfec4266c0cec..6f2379a42dd72 100755 --- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala @@ -505,6 +505,7 @@ class KafkaConfigTest { case KafkaConfig.SSLKeyManagerAlgorithmProp => case KafkaConfig.SSLTrustManagerAlgorithmProp => case KafkaConfig.SSLClientAuthProp => // ignore string + case KafkaConfig.SSLCipherSuitesProp => // ignore string case nonNegativeIntProperty => assertPropertyInvalid(getBaseProperties(), name, "not_a_number", "-1") } From 81a1f2e3dcd7fa3bf12eee2fc2743a609d734dbe Mon Sep 17 00:00:00 2001 From: Ben Stopford Date: Tue, 29 Sep 2015 19:03:13 +0100 Subject: [PATCH 2/3] KAFKA-2431: Various console and producer performance options - paramaterised test-timeout in the consumer so it's a command line option - added header for Producer/Consumer properties (so it's clear what is being overridden) - added command line option to Producer/Consumer for freeform properties of the format key=val,key=val etc - made linger.ms a command line option for ProducerPerf. This is a key setting. - cleanup based on comments from Jun/Ismael --- .../kafka/tools/ConsumerPerformance.scala | 54 ++++++++--- .../main/scala/kafka/tools/PerfConfig.scala | 6 +- .../kafka/tools/ProducerPerformance.scala | 97 ++++++++++++------- .../kafka/tools/SimplePropertiesParser.scala | 28 ++++++ .../tools/SimplePropertiesParserTest.scala | 27 ++++++ 5 files changed, 156 insertions(+), 56 deletions(-) create mode 100644 core/src/main/scala/kafka/tools/SimplePropertiesParser.scala create mode 100644 core/src/test/scala/kafka/tools/SimplePropertiesParserTest.scala diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index 907f37839d35f..2473657432650 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -19,6 +19,7 @@ package kafka.tools import java.util +import kafka.tools.SimplePropertiesParser._ import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.utils.Utils @@ -37,6 +38,9 @@ import kafka.consumer.ConsumerTimeoutException import java.text.SimpleDateFormat import java.util.concurrent.atomic.AtomicBoolean +import scala.collection.immutable +import scala.util.control.NonFatal + /** * Performance test for the full zookeeper consumer */ @@ -51,15 +55,17 @@ object ConsumerPerformance { val totalBytesRead = new AtomicLong(0) val consumerTimeout = new AtomicBoolean(false) - if (!config.hideHeader && (config.showDetailedStats || config.showAllStats)) - println("time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") + if (!config.hideHeader && (config.showDetailedStats || config.showAllStats)) { + println(s"Consumer Properties: ${config.props}\n") + println("time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") + } var startMs, endMs = 0L - if(config.useNewConsumer) { + if (config.useNewConsumer) { val consumer = new KafkaConsumer[Array[Byte], Array[Byte]](config.props) consumer.subscribe(List(config.topic)) startMs = System.currentTimeMillis - consume(consumer, List(config.topic), config.numMessages, 5000, config, totalMessagesRead, totalBytesRead) + consume(consumer, List(config.topic), config.numMessages, config.testTimeout, config, totalMessagesRead, totalBytesRead) endMs = System.currentTimeMillis consumer.close() } else { @@ -80,19 +86,18 @@ object ConsumerPerformance { thread.start for (thread <- threadList) thread.join - if(consumerTimeout.get()) + if (consumerTimeout.get()) endMs = System.currentTimeMillis - consumerConfig.consumerTimeoutMs else endMs = System.currentTimeMillis consumerConnector.shutdown() } val elapsedSecs = (endMs - startMs) / 1000.0 - val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) - - if(!config.showDetailedStats || config.showAllStats){ - if (!config.hideHeader ) + if (!config.showDetailedStats || config.showAllStats){ + if (!config.hideHeader) println("start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") + val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) println(("%s, %s, %.4f, %.4f, %d, %.4f").format(config.dateFormat.format(startMs), config.dateFormat.format(endMs), totalMBRead, totalMBRead / elapsedSecs, totalMessagesRead.get, totalMessagesRead.get / elapsedSecs)) } @@ -130,13 +135,13 @@ object ConsumerPerformance { while(messagesRead < count && System.currentTimeMillis() - lastConsumedTime <= timeout) { val records = consumer.poll(100) - if(records.count() > 0) + if (records.count() > 0) lastConsumedTime = System.currentTimeMillis for(record <- records) { messagesRead += 1 - if(record.key != null) + if (record.key != null) bytesRead += record.key.size - if(record.value != null) + if (record.value != null) bytesRead += record.value.size if (messagesRead % config.reportingInterval == 0) { @@ -208,7 +213,16 @@ object ConsumerPerformance { .describedAs("consumer properties") .ofType(classOf[java.lang.String]) .defaultsTo("") + val freeformPropsOpt = parser.accepts("consumer-properties", "Freeform consumer properties in the format key=val,key=val,key=val...") + .withRequiredArg + .describedAs("consumer properties") + .ofType(classOf[java.lang.String]) val useNewConsumerOpt = parser.accepts("new-consumer", "Use the new consumer implementation.") + val testTimeoutOpt = parser.accepts("test-timeout", "ms of inactivity after which the test will time out.") + .withRequiredArg + .describedAs("time in ms") + .ofType(classOf[java.lang.Integer]) + .defaultsTo(5000) val options = parser.parse(args: _*) @@ -217,7 +231,12 @@ object ConsumerPerformance { val useNewConsumer = options.has(useNewConsumerOpt) val props = if (options.has(propsFileOpt)) Utils.loadProps(options.valueOf(propsFileOpt)) else new Properties - if(useNewConsumer) { + + if (options.has(freeformPropsOpt)) { + props.putAll(propsFromFreeform(options.valueOf(freeformPropsOpt))) + } + + if (useNewConsumer) { import org.apache.kafka.clients.consumer.ConsumerConfig props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServersOpt)) props.put(ConsumerConfig.GROUP_ID_CONFIG, options.valueOf(groupIdOpt)) @@ -238,15 +257,18 @@ object ConsumerPerformance { props.put("num.consumer.fetchers", options.valueOf(numFetchersOpt).toString) } val numThreads = options.valueOf(numThreadsOpt).intValue + val testTimeout = options.valueOf(testTimeoutOpt).intValue val topic = options.valueOf(topicOpt) val numMessages = options.valueOf(numMessagesOpt).longValue val showDetailedStats = options.has(showDetailedStatsOpt) val showAllStats = options.has(showAllStatsOpt) val dateFormat = new SimpleDateFormat(options.valueOf(dateFormatOpt)) val hideHeader = options.has(hideHeaderOpt) - var reportingInterval = options.valueOf(reportingIntervalOpt).longValue - if(reportingInterval == -1) - reportingInterval = numMessages/10 + val reportingInterval = { + val interval = options.valueOf(reportingIntervalOpt).longValue + if (interval == -1) numMessages / 10 + else interval + } } class ConsumerPerfThread(threadId: Int, name: String, stream: KafkaStream[Array[Byte], Array[Byte]], diff --git a/core/src/main/scala/kafka/tools/PerfConfig.scala b/core/src/main/scala/kafka/tools/PerfConfig.scala index 933d0f7a075ca..713aa03aaaa09 100644 --- a/core/src/main/scala/kafka/tools/PerfConfig.scala +++ b/core/src/main/scala/kafka/tools/PerfConfig.scala @@ -26,11 +26,11 @@ class PerfConfig(args: Array[String]) { .withRequiredArg .describedAs("count") .ofType(classOf[java.lang.Long]) - val reportingIntervalOpt = parser.accepts("reporting-interval", "Interval at which to print progress info.") + val reportingIntervalOpt = parser.accepts("reporting-interval", "Interval at which to print progress info. If not supplied this is set dynamically") .withRequiredArg - .describedAs("size") + .describedAs("number of messages after which we print progress") .ofType(classOf[java.lang.Integer]) - .defaultsTo(-1) + .defaultsTo(-1)//this default is overridden dynamically based on the supplied batch size val dateFormatOpt = parser.accepts("date-format", "The date format to use for formatting the time field. " + "See java.text.SimpleDateFormat for options.") .withRequiredArg diff --git a/core/src/main/scala/kafka/tools/ProducerPerformance.scala b/core/src/main/scala/kafka/tools/ProducerPerformance.scala index 5f718bcbc9859..dbeff14bd1ea6 100644 --- a/core/src/main/scala/kafka/tools/ProducerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ProducerPerformance.scala @@ -19,6 +19,7 @@ package kafka.tools import kafka.metrics.KafkaMetricsReporter import kafka.producer.{OldProducer, NewShinyProducer} +import kafka.tools.SimplePropertiesParser._ import kafka.utils.{ToolsUtils, VerifiableProperties, Logging, CommandLineUtils} import kafka.message.CompressionCodec import kafka.serializer._ @@ -56,12 +57,16 @@ object ProducerPerformance extends Logging { val startMs = System.currentTimeMillis val rand = new java.util.Random - if (!config.hideHeader) + val props: Properties = properties(config) + + if (!config.hideHeader) { + println(s"Producer Properties: ${props}\n") println("start.time, end.time, compression, message.size, batch.size, total.data.sent.in.MB, MB.sec, " + "total.data.sent.in.nMsg, nMsg.sec") + } for (i <- 0 until config.numThreads) { - executor.execute(new ProducerThread(i, config, totalBytesSent, totalMessagesSent, allDone, rand)) + executor.execute(new ProducerThread(i, config, totalBytesSent, totalMessagesSent, allDone, rand, props)) } allDone.await() @@ -74,6 +79,42 @@ object ProducerPerformance extends Logging { totalMBSent / elapsedSecs, totalMessagesSent.get, totalMessagesSent.get / elapsedSecs)) } + def properties(config: ProducerPerfConfig): Properties = { + val props = new Properties() + if (config.useNewProducer) { + import org.apache.kafka.clients.producer.ProducerConfig + props.putAll(config.producerProps) + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.brokerList) + props.put(ProducerConfig.SEND_BUFFER_CONFIG, (64 * 1024).toString) + props.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-performance") + props.put(ProducerConfig.ACKS_CONFIG, config.producerRequestRequiredAcks.toString) + props.put(ProducerConfig.RETRIES_CONFIG, config.producerNumRetries.toString) + props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, config.producerRetryBackoffMs.toString) + props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, config.compressionCodec.name) + if (config.lingerTime != null) props.put(ProducerConfig.LINGER_MS_CONFIG, config.lingerTime) + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer") + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer") + } else { + props.putAll(config.producerProps) + props.put("metadata.broker.list", config.brokerList) + props.put("compression.codec", config.compressionCodec.codec.toString) + props.put("send.buffer.bytes", (64 * 1024).toString) + if (!config.isSync) { + props.put("producer.type", "async") + props.put("batch.num.messages", config.batchSize.toString) + props.put("queue.enqueue.timeout.ms", "-1") + } + props.put("client.id", "producer-performance") + props.put("request.required.acks", config.producerRequestRequiredAcks.toString) + props.put("request.timeout.ms", config.producerRequestTimeoutMs.toString) + props.put("message.send.max.retries", config.producerNumRetries.toString) + props.put("retry.backoff.ms", config.producerRetryBackoffMs.toString) + props.put("serializer.class", classOf[DefaultEncoder].getName) + props.put("key.serializer.class", classOf[NullEncoder[Long]].getName) + } + props + } + class ProducerPerfConfig(args: Array[String]) extends PerfConfig(args) { val brokerListOpt = parser.accepts("broker-list", "REQUIRED: broker info the list of broker host and port for bootstrap.") .withRequiredArg @@ -111,6 +152,10 @@ object ProducerPerformance extends Logging { .describedAs("number of threads") .ofType(classOf[java.lang.Integer]) .defaultsTo(1) + val lingerTimeOpt = parser.accepts("linger-ms", "This setting adds a small amount of artificial delay to writes to allow them to batch. See documentation for linger.ms") + .withRequiredArg + .describedAs("ms of linger") + .ofType(classOf[java.lang.Integer]) val initialMessageIdOpt = parser.accepts("initial-message-id", "The is used for generating test data, If set, messages will be tagged with an " + "ID and sent by producer starting from this ID sequentially. Message content will be String type and " + "in the form of 'Message:000...1:xxx...'") @@ -129,6 +174,10 @@ object ProducerPerformance extends Logging { .describedAs("metrics directory") .ofType(classOf[java.lang.String]) val useNewProducerOpt = parser.accepts("new-producer", "Use the new producer implementation.") + val freeformPropsOpt = parser.accepts("producer-properties", "Freeform set of producer properties in the format name=value,name=value...") + .withRequiredArg() + .describedAs("freeform properties") + .ofType(classOf[String]) val options = parser.parse(args: _*) CommandLineUtils.checkRequiredArgs(parser, options, topicsOpt, brokerListOpt, numMessagesOpt) @@ -146,6 +195,7 @@ object ProducerPerformance extends Logging { var isSync = options.has(syncOpt) var batchSize = options.valueOf(batchSizeOpt).intValue var numThreads = options.valueOf(numThreadsOpt).intValue + val lingerTime = options.valueOf(lingerTimeOpt) val compressionCodec = CompressionCodec.getCompressionCodec(options.valueOf(compressionCodecOpt).intValue) val seqIdMode = options.has(initialMessageIdOpt) var initialMessageId: Int = 0 @@ -164,6 +214,10 @@ object ProducerPerformance extends Logging { else new Properties() + if (options.has(freeformPropsOpt)) { + producerProps.putAll(propsFromFreeform(options.valueOf(freeformPropsOpt))) + } + if (csvMetricsReporterEnabled) { val props = new Properties() props.put("kafka.metrics.polling.interval.secs", "1") @@ -185,45 +239,14 @@ object ProducerPerformance extends Logging { val totalBytesSent: AtomicLong, val totalMessagesSent: AtomicLong, val allDone: CountDownLatch, - val rand: Random) extends Runnable { + val rand: Random, + val props: Properties) extends Runnable { val seqIdNumDigit = 10 // no. of digits for max int value val messagesPerThread = config.numMessages / config.numThreads debug("Messages per thread = " + messagesPerThread) - val props = new Properties() - val producer = - if (config.useNewProducer) { - import org.apache.kafka.clients.producer.ProducerConfig - props.putAll(config.producerProps) - props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.brokerList) - props.put(ProducerConfig.SEND_BUFFER_CONFIG, (64 * 1024).toString) - props.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-performance") - props.put(ProducerConfig.ACKS_CONFIG, config.producerRequestRequiredAcks.toString) - props.put(ProducerConfig.RETRIES_CONFIG, config.producerNumRetries.toString) - props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, config.producerRetryBackoffMs.toString) - props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, config.compressionCodec.name) - props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer") - props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer") - new NewShinyProducer(props) - } else { - props.putAll(config.producerProps) - props.put("metadata.broker.list", config.brokerList) - props.put("compression.codec", config.compressionCodec.codec.toString) - props.put("send.buffer.bytes", (64 * 1024).toString) - if (!config.isSync) { - props.put("producer.type", "async") - props.put("batch.num.messages", config.batchSize.toString) - props.put("queue.enqueue.timeout.ms", "-1") - } - props.put("client.id", "producer-performance") - props.put("request.required.acks", config.producerRequestRequiredAcks.toString) - props.put("request.timeout.ms", config.producerRequestTimeoutMs.toString) - props.put("message.send.max.retries", config.producerNumRetries.toString) - props.put("retry.backoff.ms", config.producerRetryBackoffMs.toString) - props.put("serializer.class", classOf[DefaultEncoder].getName) - props.put("key.serializer.class", classOf[NullEncoder[Long]].getName) - new OldProducer(props) - } + + val producer = if (config.useNewProducer) new NewShinyProducer(props) else new OldProducer(props) // generate the sequential message ID private val SEP = ":" // message field separator diff --git a/core/src/main/scala/kafka/tools/SimplePropertiesParser.scala b/core/src/main/scala/kafka/tools/SimplePropertiesParser.scala new file mode 100644 index 0000000000000..978d905ad7e02 --- /dev/null +++ b/core/src/main/scala/kafka/tools/SimplePropertiesParser.scala @@ -0,0 +1,28 @@ +package kafka.tools + +import scala.collection.JavaConverters._ +import scala.collection.immutable +import scala.util.control.NonFatal + +object SimplePropertiesParser { + + /** + * Take a simple string of the form key=val,key=val,key=val and parse into a map. Useful for command line + * arguments. + * @param freeform String of format "key=val,key=val,key=val" + * @return + */ + def propsFromFreeform(freeform: String): java.util.Map[String, String] = { + val argProps: immutable.Map[String, String] = freeform.split(",") + .map(_.split("=")) + .map { + case Array(k, v) => (k, v) + case _ => exception(freeform) + }.toMap + argProps.asJava + } + + def exception(freeform: String): Nothing = { + throw new IllegalArgumentException("There was a problem parsing freeform properties string. These should be of the form key=val,key=val... Input was: " + freeform) + } +} diff --git a/core/src/test/scala/kafka/tools/SimplePropertiesParserTest.scala b/core/src/test/scala/kafka/tools/SimplePropertiesParserTest.scala new file mode 100644 index 0000000000000..4db7578b06a13 --- /dev/null +++ b/core/src/test/scala/kafka/tools/SimplePropertiesParserTest.scala @@ -0,0 +1,27 @@ +package kafka.tools + +import java.util + +import org.junit.Assert +import org.junit.Test + +class SimplePropertiesParserTest { + + @Test + def shouldParseGoodProperties: Unit = { + val result: util.Map[String, String] = SimplePropertiesParser.propsFromFreeform("key1=val1,key2=val2") + assert(result.size() == 2) + assert(result.get("key1") == "val1") + assert(result.get("key2") == "val2") + } + + @Test + def shouldFailOnBadProperties: Unit = { + try { + SimplePropertiesParser.propsFromFreeform("key1=val1,key2") + Assert.fail("Should have thrown an exception") + } catch { + case e: IllegalArgumentException => // expected exception + } + } +} From 149ee434ea1353a85c903965f2d05bc705b3a171 Mon Sep 17 00:00:00 2001 From: Ben Stopford Date: Tue, 29 Sep 2015 22:16:31 +0100 Subject: [PATCH 3/3] KAFKA-2431: added missing license files --- .../kafka/tools/SimplePropertiesParser.scala | 16 ++++++++++++++++ .../kafka/tools/SimplePropertiesParserTest.scala | 16 ++++++++++++++++ 2 files changed, 32 insertions(+) diff --git a/core/src/main/scala/kafka/tools/SimplePropertiesParser.scala b/core/src/main/scala/kafka/tools/SimplePropertiesParser.scala index 978d905ad7e02..0883b0039ab49 100644 --- a/core/src/main/scala/kafka/tools/SimplePropertiesParser.scala +++ b/core/src/main/scala/kafka/tools/SimplePropertiesParser.scala @@ -1,3 +1,19 @@ +/** + * 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 kafka.tools import scala.collection.JavaConverters._ diff --git a/core/src/test/scala/kafka/tools/SimplePropertiesParserTest.scala b/core/src/test/scala/kafka/tools/SimplePropertiesParserTest.scala index 4db7578b06a13..f3d942c538fcf 100644 --- a/core/src/test/scala/kafka/tools/SimplePropertiesParserTest.scala +++ b/core/src/test/scala/kafka/tools/SimplePropertiesParserTest.scala @@ -1,3 +1,19 @@ +/** + * 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 kafka.tools import java.util