Skip to content
Closed
Show file tree
Hide file tree
Changes from 15 commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
6a1ca42
Initial pass, main code compiles
fpj Oct 12, 2015
afeafab
Changes to tests to accomodate the refactoring of ZkUtils.
fpj Oct 13, 2015
66b116a
Removed whitespaces.
fpj Oct 13, 2015
36c3720
KAFKA-2639: Added close() method to ZkUtils
fpj Oct 13, 2015
a7ae433
KAFKA-2639: Fixed PartitionAssignorTest
fpj Oct 13, 2015
78ee23d
KAFKA-2639: Fixed ReplicaManagerTest
fpj Oct 13, 2015
2e888de
KAFKA-2639: Fixed ZKPathTest
fpj Oct 13, 2015
b94fd4b
KAFKA-2639: Made isSecure a parameter of the factory methods for ZkUtils
fpj Oct 13, 2015
bd46f61
KAFKA-2639: Fixed KafkaConfigTest.
fpj Oct 13, 2015
8c69f23
KAFKA-2639: Removing config via KafkaConfig.
fpj Oct 14, 2015
00a8169
KAFKA-2639: Removed whitespaces.
fpj Oct 14, 2015
311612f
KAFKA-2639: Removed unrelated comment from ZkUtils.
fpj Oct 14, 2015
fb9a52a
KAFKA-2639: Moved isSecure to JaasUtils in clients.
fpj Oct 14, 2015
8314c7f
KAFKA-2639: Covering more zk system properties.
fpj Oct 14, 2015
76a802d
KAFKA-2639: Small update to log message and exception message in Jaas…
fpj Oct 14, 2015
fd799b0
Merge remote-tracking branch 'upstream/trunk' into KAFKA-2639
fpj Oct 15, 2015
58e17b2
KAFKA-2639: create->apply and isSecure->isZkSecurityEnabled
fpj Oct 15, 2015
d79e108
KAFKA-2639: Various changes due to comments.
fpj Oct 18, 2015
7fb8ffa
KAFKA-2639: Second round of comments, mainly indentation.
fpj Oct 18, 2015
15b7b52
KAFKA-2639: Fixed more indentatin issues.
fpj Oct 18, 2015
91c0028
KAFKA-2639: Fixed more indentation issues.
fpj Oct 18, 2015
d788ce5
Merge branch 'KAFKA-2639' of https://github.com/fpj/kafka into KAFKA-…
fpj Oct 18, 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
1 change: 1 addition & 0 deletions checkstyle/import-control.xml
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
<allow pkg="org.powermock" />

<allow pkg="javax.net.ssl" />
<allow pkg="javax.security.auth" />

<!-- no one depends on the server -->
<disallow pkg="kafka" />
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
/**
* 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.kafka.common.security;

import java.io.File;
import java.net.URI;
import java.security.URIParameter;
import javax.security.auth.login.Configuration;
import org.apache.kafka.common.KafkaException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class JaasUtils {
private static final Logger LOG = LoggerFactory.getLogger(JaasUtils.class);
public static final String LOGIN_CONTEXT_SERVER = "KafkaServer";
public static final String LOGIN_CONTEXT_CLIENT = "KafkaClient";
public static final String SERVICE_NAME = "serviceName";
public static final String JAVA_LOGIN_CONFIG_PARAM = "java.security.auth.login.config";
public static final String ZK_SASL_CLIENT = "zookeeper.sasl.client";

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 thought on the broker side, we want to add a broker config to enable setting ZK acl. On the client side (this will mostly be for tools), we will just enable ZK acl if the system property "java.security.auth.login.config" is set.

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.

The config parameter is introduced in the PR #313. Here I'm focusing on the refactoring and I just followed the approach in the original PR #93 of using the system property.

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.

This sounds good.

public static final String ZK_LOGIN_CONTEXT_NAME_KEY = "zookeeper.sasl.clientconfig";

public static boolean isSecure(String loginConfigFile) {

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.

As this is ZK-specific, the method name should make that clear.

boolean isSecurityEnabled = false;
boolean zkSaslEnabled = Boolean.getBoolean(System.getProperty(ZK_SASL_CLIENT, "true"));
String zkLoginContextName = System.getProperty(ZK_LOGIN_CONTEXT_NAME_KEY, "Client");

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.

Is the ZK login context name customizable? I thought Zookeeper client library always expects it to be Client.

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 is customizable. This is not properly documented in ZK.



if (loginConfigFile != null && loginConfigFile.length() > 0) {
File configFile = new File(loginConfigFile);
if (!configFile.canRead()) {
throw new KafkaException("File " + loginConfigFile + "cannot be read.");
}
try {
URI configUri = configFile.toURI();
Configuration loginConf = Configuration.getInstance("JavaLoginConfig", new URIParameter(configUri));
isSecurityEnabled = loginConf.getAppConfigurationEntry(zkLoginContextName) != null;
} catch (Exception e) {
throw new KafkaException(e);
}
if (isSecurityEnabled && !zkSaslEnabled) {
LOG.error("JAAS file is present, but system property " +
ZK_SASL_CLIENT + " is set to false, which disables " +
"SASL in the ZooKeeper client");
throw new KafkaException("Exception while determining if ZooKeeper is secure");

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.

I had a look at instances where we call LOG.error or error in the Kafka codebase to see what our general error-handling strategy (log and throw or just throw). It seems to me that in the majority of the cases we either log or throw. But there are cases where we do both. And the Copycat code seems to do both quite regularly.

As we discussed, I prefer just throw to avoid duplicate entries in the logs and because it's more composable (a utility method like this can be called from multiple contexts and it's better to let the caller decide how to handle the exception).

I would be interested in @junrao's opinion on this.

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.

@ijuma I was thinking that we can't guarantee that exceptions are always logged. They can be caught and they might not end up in the log output. The log error message is guaranteed to be in the log output and to indicate that an error occurred.

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.

Yeah, that's the argument for having it there (thanks for mentioning it). However, if the caller can actually handle the exception, then it is perhaps not something that should be logged at error level anymore.

Whatever approach we take generally, it is important to be consistent because it breaks down if we use a mixture of approaches (in this method, we are only logging in only one of the exception paths, for example). This is why I asked for @junrao's input.

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.

In general, if the caller is already logging the exception, we don't have to log it when the exception is thrown, unless there is more context when the exception is thrown that needs to be logged.

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.

Actually, do we need to throw exception here? During the broker migration, it's ok to set the jaas file, but disable setting the acl.

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.

This isn't used during migration only @junrao. The code for migration is inPR #313.

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.

Ok, we can revisit this in PR #313.

}
}

return isSecurityEnabled;
}
}
128 changes: 64 additions & 64 deletions core/src/main/scala/kafka/admin/AdminUtils.scala

Large diffs are not rendered by default.

26 changes: 15 additions & 11 deletions core/src/main/scala/kafka/admin/ConfigCommand.scala
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import org.I0Itec.zkclient.ZkClient
import scala.collection._
import scala.collection.JavaConversions._
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.common.security.JaasUtils


/**
Expand All @@ -42,52 +43,55 @@ object ConfigCommand {

opts.checkArgs()

val zkClient = ZkUtils.createZkClient(opts.options.valueOf(opts.zkConnectOpt), 30000, 30000)
val zkUtils = ZkUtils.create(opts.options.valueOf(opts.zkConnectOpt),
30000,
30000,
JaasUtils.isSecure(System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM)))

try {
if (opts.options.has(opts.alterOpt))
alterConfig(zkClient, opts)
alterConfig(zkUtils, opts)
else if (opts.options.has(opts.describeOpt))
describeConfig(zkClient, opts)
describeConfig(zkUtils, opts)
} catch {
case e: Throwable =>
println("Error while executing topic command " + e.getMessage)
println(Utils.stackTrace(e))
} finally {
zkClient.close()
zkUtils.close()
}
}

private def alterConfig(zkClient: ZkClient, opts: ConfigCommandOptions) {
private def alterConfig(zkUtils: ZkUtils, opts: ConfigCommandOptions) {
val configsToBeAdded = parseConfigsToBeAdded(opts)
val configsToBeDeleted = parseConfigsToBeDeleted(opts)
val entityType = opts.options.valueOf(opts.entityType)
val entityName = opts.options.valueOf(opts.entityName)

// compile the final set of configs
val configs = AdminUtils.fetchEntityConfig(zkClient, entityType, entityName)
val configs = AdminUtils.fetchEntityConfig(zkUtils, entityType, entityName)
configs.putAll(configsToBeAdded)
configsToBeDeleted.foreach(config => configs.remove(config))

if (entityType.equals(ConfigType.Topic)) {
AdminUtils.changeTopicConfig(zkClient, entityName, configs)
AdminUtils.changeTopicConfig(zkUtils, entityName, configs)
println("Updated config for topic: \"%s\".".format(entityName))
} else {
AdminUtils.changeClientIdConfig(zkClient, entityName, configs)
AdminUtils.changeClientIdConfig(zkUtils, entityName, configs)
println("Updated config for clientId: \"%s\".".format(entityName))
}
}

private def describeConfig(zkClient: ZkClient, opts: ConfigCommandOptions) {
private def describeConfig(zkUtils: ZkUtils, opts: ConfigCommandOptions) {
val entityType = opts.options.valueOf(opts.entityType)
val entityNames: Seq[String] =
if (opts.options.has(opts.entityName))
Seq(opts.options.valueOf(opts.entityName))
else
ZkUtils.getAllEntitiesWithConfig(zkClient, entityType)
zkUtils.getAllEntitiesWithConfig(entityType)

for (entityName <- entityNames) {
val configs = AdminUtils.fetchEntityConfig(zkClient, entityType, entityName)
val configs = AdminUtils.fetchEntityConfig(zkUtils, entityType, entityName)
println("Configs for %s:%s are %s"
.format(entityType, entityName, configs.map(kv => kv._1 + "=" + kv._2).mkString(",")))
}
Expand Down
74 changes: 39 additions & 35 deletions core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ import scala.collection.{Set, mutable}
import kafka.consumer.SimpleConsumer
import collection.JavaConversions._
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.common.security.JaasUtils


object ConsumerGroupCommand {
Expand All @@ -48,57 +49,60 @@ object ConsumerGroupCommand {

opts.checkArgs()

val zkClient = ZkUtils.createZkClient(opts.options.valueOf(opts.zkConnectOpt), 30000, 30000)
val zkUtils = ZkUtils.create(opts.options.valueOf(opts.zkConnectOpt),
30000,
30000,
JaasUtils.isSecure(System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM)))

try {
if (opts.options.has(opts.listOpt))
list(zkClient)
list(zkUtils)
else if (opts.options.has(opts.describeOpt))
describe(zkClient, opts)
describe(zkUtils, opts)
else if (opts.options.has(opts.deleteOpt))
delete(zkClient, opts)
delete(zkUtils, opts)
} catch {
case e: Throwable =>
println("Error while executing consumer group command " + e.getMessage)
println(Utils.stackTrace(e))
} finally {
zkClient.close()
zkUtils.close()
}
}

def list(zkClient: ZkClient) {
ZkUtils.getConsumerGroups(zkClient).foreach(println)
def list(zkUtils: ZkUtils) {
zkUtils.getConsumerGroups().foreach(println)
}

def describe(zkClient: ZkClient, opts: ConsumerGroupCommandOptions) {
def describe(zkUtils: ZkUtils, opts: ConsumerGroupCommandOptions) {
val configs = parseConfigs(opts)
val channelSocketTimeoutMs = configs.getProperty("channelSocketTimeoutMs", "600").toInt
val channelRetryBackoffMs = configs.getProperty("channelRetryBackoffMsOpt", "300").toInt
val group = opts.options.valueOf(opts.groupOpt)
val topics = ZkUtils.getTopicsByConsumerGroup(zkClient, group)
val topics = zkUtils.getTopicsByConsumerGroup(group)
if (topics.isEmpty) {
println("No topic available for consumer group provided")
}
topics.foreach(topic => describeTopic(zkClient, group, topic, channelSocketTimeoutMs, channelRetryBackoffMs))
topics.foreach(topic => describeTopic(zkUtils, group, topic, channelSocketTimeoutMs, channelRetryBackoffMs))
}

def delete(zkClient: ZkClient, opts: ConsumerGroupCommandOptions) {
def delete(zkUtils: ZkUtils, opts: ConsumerGroupCommandOptions) {
if (opts.options.has(opts.groupOpt) && opts.options.has(opts.topicOpt)) {
deleteForTopic(zkClient, opts)
deleteForTopic(zkUtils, opts)
}
else if (opts.options.has(opts.groupOpt)) {
deleteForGroup(zkClient, opts)
deleteForGroup(zkUtils, opts)
}
else if (opts.options.has(opts.topicOpt)) {
deleteAllForTopic(zkClient, opts)
deleteAllForTopic(zkUtils, opts)
}
}

private def deleteForGroup(zkClient: ZkClient, opts: ConsumerGroupCommandOptions) {
private def deleteForGroup(zkUtils: ZkUtils, opts: ConsumerGroupCommandOptions) {
val groups = opts.options.valuesOf(opts.groupOpt)
groups.foreach { group =>
try {
if (AdminUtils.deleteConsumerGroupInZK(zkClient, group))
if (AdminUtils.deleteConsumerGroupInZK(zkUtils, group))
println("Deleted all consumer group information for group %s in zookeeper.".format(group))
else
println("Delete for group %s failed because its consumers are still active.".format(group))
Expand All @@ -110,13 +114,13 @@ object ConsumerGroupCommand {
}
}

private def deleteForTopic(zkClient: ZkClient, opts: ConsumerGroupCommandOptions) {
private def deleteForTopic(zkUtils: ZkUtils, opts: ConsumerGroupCommandOptions) {
val groups = opts.options.valuesOf(opts.groupOpt)
val topic = opts.options.valueOf(opts.topicOpt)
Topic.validate(topic)
groups.foreach { group =>
try {
if (AdminUtils.deleteConsumerGroupInfoForTopicInZK(zkClient, group, topic))
if (AdminUtils.deleteConsumerGroupInfoForTopicInZK(zkUtils, group, topic))
println("Deleted consumer group information for group %s topic %s in zookeeper.".format(group, topic))
else
println("Delete for group %s topic %s failed because its consumers are still active.".format(group, topic))
Expand All @@ -128,10 +132,10 @@ object ConsumerGroupCommand {
}
}

private def deleteAllForTopic(zkClient: ZkClient, opts: ConsumerGroupCommandOptions) {
private def deleteAllForTopic(zkUtils: ZkUtils, opts: ConsumerGroupCommandOptions) {
val topic = opts.options.valueOf(opts.topicOpt)
Topic.validate(topic)
AdminUtils.deleteAllConsumerGroupInfoForTopicInZK(zkClient, topic)
AdminUtils.deleteAllConsumerGroupInfoForTopicInZK(zkUtils, topic)
println("Deleted consumer group information for all inactive consumer groups for topic %s in zookeeper.".format(topic))
}

Expand All @@ -144,35 +148,35 @@ object ConsumerGroupCommand {
props
}

private def describeTopic(zkClient: ZkClient,
private def describeTopic(zkUtils: ZkUtils,
group: String,
topic: String,
channelSocketTimeoutMs: Int,
channelRetryBackoffMs: Int) {
val topicPartitions = getTopicPartitions(zkClient, topic)
val partitionOffsets = getPartitionOffsets(zkClient, group, topicPartitions, channelSocketTimeoutMs, channelRetryBackoffMs)
val topicPartitions = getTopicPartitions(zkUtils, topic)
val partitionOffsets = getPartitionOffsets(zkUtils, group, topicPartitions, channelSocketTimeoutMs, channelRetryBackoffMs)
println("%s, %s, %s, %s, %s, %s, %s"
.format("GROUP", "TOPIC", "PARTITION", "CURRENT OFFSET", "LOG END OFFSET", "LAG", "OWNER"))
topicPartitions
.sortBy { case topicPartition => topicPartition.partition }
.foreach { topicPartition =>
describePartition(zkClient, group, topicPartition.topic, topicPartition.partition, partitionOffsets.get(topicPartition))
describePartition(zkUtils, group, topicPartition.topic, topicPartition.partition, partitionOffsets.get(topicPartition))
}
}

private def getTopicPartitions(zkClient: ZkClient, topic: String) = {
val topicPartitionMap = ZkUtils.getPartitionsForTopics(zkClient, Seq(topic))
private def getTopicPartitions(zkUtils: ZkUtils, topic: String) = {
val topicPartitionMap = zkUtils.getPartitionsForTopics(Seq(topic))
val partitions = topicPartitionMap.getOrElse(topic, Seq.empty)
partitions.map(TopicAndPartition(topic, _))
}

private def getPartitionOffsets(zkClient: ZkClient,
private def getPartitionOffsets(zkUtils: ZkUtils,
group: String,
topicPartitions: Seq[TopicAndPartition],
channelSocketTimeoutMs: Int,
channelRetryBackoffMs: Int): Map[TopicAndPartition, Long] = {
val offsetMap = mutable.Map[TopicAndPartition, Long]()
val channel = ClientUtils.channelToOffsetManager(group, zkClient, channelSocketTimeoutMs, channelRetryBackoffMs)
val channel = ClientUtils.channelToOffsetManager(group, zkUtils, channelSocketTimeoutMs, channelRetryBackoffMs)
channel.send(OffsetFetchRequest(group, topicPartitions))
val offsetFetchResponse = OffsetFetchResponse.readFrom(channel.receive().payload())

Expand All @@ -182,7 +186,7 @@ object ConsumerGroupCommand {
// this group may not have migrated off zookeeper for offsets storage (we don't expose the dual-commit option in this tool
// (meaning the lag may be off until all the consumers in the group have the same setting for offsets storage)
try {
val offset = ZkUtils.readData(zkClient, topicDirs.consumerOffsetDir + "/" + topicAndPartition.partition)._1.toLong
val offset = zkUtils.readData(topicDirs.consumerOffsetDir + "/" + topicAndPartition.partition)._1.toLong
offsetMap.put(topicAndPartition, offset)
} catch {
case z: ZkNoNodeException =>
Expand All @@ -200,20 +204,20 @@ object ConsumerGroupCommand {
offsetMap.toMap
}

private def describePartition(zkClient: ZkClient,
private def describePartition(zkUtils: ZkUtils,
group: String,
topic: String,
partition: Int,
offsetOpt: Option[Long]) {
val topicAndPartition = TopicAndPartition(topic, partition)
val groupDirs = new ZKGroupTopicDirs(group, topic)
val owner = ZkUtils.readDataMaybeNull(zkClient, groupDirs.consumerOwnerDir + "/" + partition)._1
ZkUtils.getLeaderForPartition(zkClient, topic, partition) match {
val owner = zkUtils.readDataMaybeNull(groupDirs.consumerOwnerDir + "/" + partition)._1
zkUtils.getLeaderForPartition(topic, partition) match {
case Some(-1) =>
println("%s, %s, %s, %s, %s, %s, %s"
.format(group, topic, partition, offsetOpt.getOrElse("unknown"), "unknown", "unknown", owner.getOrElse("none")))
case Some(brokerId) =>
val consumerOpt = getConsumer(zkClient, brokerId)
val consumerOpt = getConsumer(zkUtils, brokerId)
consumerOpt match {
case Some(consumer) =>
val request =
Expand All @@ -231,9 +235,9 @@ object ConsumerGroupCommand {
}
}

private def getConsumer(zkClient: ZkClient, brokerId: Int): Option[SimpleConsumer] = {
private def getConsumer(zkUtils: ZkUtils, brokerId: Int): Option[SimpleConsumer] = {
try {
ZkUtils.readDataMaybeNull(zkClient, ZkUtils.BrokerIdsPath + "/" + brokerId)._1 match {
zkUtils.readDataMaybeNull(ZkUtils.BrokerIdsPath + "/" + brokerId)._1 match {
case Some(brokerInfoString) =>
Json.parseFull(brokerInfoString) match {
case Some(m) =>
Expand Down
Loading