Skip to content

KAFKA-2431: Easier Testing of SSL #217

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
benstopford wants to merge 3 commits into from
Closed

KAFKA-2431: Easier Testing of SSL #217

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
cd4fda6
KAFKA-2431: Ease testing of SSL
benstopford Sep 8, 2015
81a1f2e
KAFKA-2431: Various console and producer performance options
benstopford Sep 29, 2015
149ee43
KAFKA-2431: added missing license files
benstopford Sep 29, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/config/SSLConfigs.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down
5 changes: 3 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,12 +60,13 @@ public void configure(Map<String, ?> configs) throws KafkaException {

if (configs.get(SSLConfigs.SSL_CIPHER_SUITES_CONFIG) != null) {
List<String> cipherSuitesList = (List<String>) 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()]);

@ijuma ijuma Sep 16, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Out of curiosity, why are we doing the isEmpty check here? We don't do the same check when setting the enabledProtocols. It seems to add noise for little benefit given how this is used.

@benstopford benstopford Sep 19, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's because the enabled protocols have a kafka supplied default value. With the cipher suite I left it up to the JVM.

@ijuma ijuma Sep 20, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh I see, if you pass an empty collection, the JDK defaults are no longer used? The change makes sense then. We should probably be using the JDK defaults for enabled protocols by default too IMO, but that can be handled via KAFKA-2456

@benstopford benstopford Sep 21, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes - I agree. I don't think overriding the JVM defaults is a good idea unless there is a specific, documented intent.

}

if (configs.get(SSLConfigs.SSL_ENABLED_PROTOCOLS_CONFIG) != null) {
List<String> enabledProtocolsList = (List<String>) 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)) {
Expand Down
4 changes: 4 additions & 0 deletions core/src/main/scala/kafka/server/KafkaConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,6 +176,7 @@ object Defaults {
val SSLClientAuthRequested = "requested"
val SSLClientAuthNone = "none"
val SSLClientAuth = SSLClientAuthNone
val SSLCipherSuites = ""

}

Expand Down Expand Up @@ -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)

}

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
}

Expand Down
67 changes: 49 additions & 18 deletions core/src/main/scala/kafka/tools/ConsumerPerformance.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,9 @@ package kafka.tools

import java.util

import kafka.tools.SimplePropertiesParser._
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.utils.Utils

import scala.collection.JavaConversions._
import java.util.concurrent.atomic.AtomicLong
Expand All @@ -36,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
*/
Expand All @@ -50,19 +55,17 @@ 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
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, 1000, config, totalMessagesRead, totalBytesRead)
consume(consumer, List(config.topic), config.numMessages, config.testTimeout, config, totalMessagesRead, totalBytesRead)
endMs = System.currentTimeMillis
consumer.close()
} else {
Expand All @@ -83,14 +86,17 @@ 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
if (!config.showDetailedStats) {
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))
Expand Down Expand Up @@ -129,17 +135,17 @@ 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) {
if (config.showDetailedStats)
if (config.showDetailedStats || config.showAllStats)
printProgressMessage(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, lastReportTime, System.currentTimeMillis, config.dateFormat)
lastReportTime = System.currentTimeMillis
lastMessagesRead = messagesRead
Expand Down Expand Up @@ -201,17 +207,36 @@ 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 freeformPropsOpt = parser.accepts("consumer-properties", "Freeform consumer properties in the format key=val,key=val,key=val...")

@junrao junrao Oct 2, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need both consumer-properties and consumer.config? It seems that having one of the two is enough? If we want to keep both, perhaps we should do the same in ProducerPerformance.

.withRequiredArg
.describedAs("consumer properties")
.ofType(classOf[java.lang.String])

@ijuma ijuma Oct 2, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just noticed that ConsoleProducer has the following:

    val producerPropertyOpt = parser.accepts("producer-property", "A mechanism to pass user-defined properties in the form key=value to the producer. ")
            .withRequiredArg
            .describedAs("producer_prop")
            .ofType(classOf[String])

And it uses CommandLineUtils.parseKeyValueArgs(options.valuesOf(propertyOpt)). We probably should be using the same naming and parsing code here.

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: _*)

CommandLineUtils.checkRequiredArgs(parser, options, topicOpt, numMessagesOpt)

val useNewConsumer = options.has(useNewConsumerOpt)

val props = new Properties
if(useNewConsumer) {
val props = if (options.has(propsFileOpt)) Utils.loadProps(options.valueOf(propsFileOpt)) else new Properties

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))
Expand All @@ -232,12 +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 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)
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]],
Expand All @@ -260,7 +291,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
Expand Down
8 changes: 5 additions & 3 deletions core/src/main/scala/kafka/tools/PerfConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(5000)
.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
Expand All @@ -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 " +

@ijuma ijuma Sep 24, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we really need a new option for this? I was a bit surprised that showDetailedStats doesn't show the final result to be honest. You probably did this for compatibility reasons, but it would be good to get a second opinion from @gwenshap or @junrao.

@benstopford benstopford Sep 29, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I know exactly what you mean fella. In fact my first reaction was to just change the behaviour, but it's an external interface so people may have coupled to this behaviour. That's why I went down the additional option route.

@ijuma ijuma Sep 29, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I understand. It seems like we don't offer strong compatibility guarantees for these scripts though, see discussion in #242. Do you think that people would actually rely on the fact that there is no additional entry with the overall results at the end?

@benstopford benstopford Sep 29, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Quite likely not I expect, but we should overhaul these scripts if we're going down that route. I should add that the way we use jopt a little problematic. We end up overriding the system's default values just because jopt leads you down the line of providing a default value. Lots of the values used don't make much sense. For settings that map directly to kafka properties it would be better if we enforced that mapping explicitly.

@junrao junrao Oct 2, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree with @ijuma . It seems it's simpler to just always print the final result instead of introducing a new option. This is a change of behavior, but it probably matches most people's expectation. ProducerPerformance is already like that.

"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
Expand Down
Loading