Skip to content

Kafka 3078 #747

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
SinghAsDev wants to merge 1 commit into from
Closed

Kafka 3078 #747

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
12 changes: 10 additions & 2 deletions tests/kafkatest/services/kafka_log4j_appender.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,10 +49,18 @@ def start_cmd(self, node):

if self.max_messages > 0:
cmd += " --max-messages %s" % str(self.max_messages)
if self.security_protocol == SecurityConfig.SSL:
cmd += " --security-protocol SSL"
if self.security_protocol != SecurityConfig.PLAINTEXT:
cmd += " --security-protocol %s" % str(self.security_protocol)
if self.security_protocol == SecurityConfig.SSL or self.security_protocol == SecurityConfig.SASL_SSL:
cmd += " --ssl-truststore-location %s" % str(SecurityConfig.TRUSTSTORE_PATH)
cmd += " --ssl-truststore-password %s" % str(SecurityConfig.ssl_stores['ssl.truststore.password'])
if self.security_protocol == SecurityConfig.SASL_PLAINTEXT or \

@ewencp ewencp Jan 12, 2016

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.

Not a big deal, but for future reference doing something like if self.security_protocol in [SecurityConfig.SASL_PLAINTEXT, SecurityConfig.SASL_SSL, SecurityConfig.SASL_MECHANISM_GSSAPI, SecurityConfig.SASL_MECHANISM_PLAIN] is a more concise way to express conditions like this. In fact, it might be valuable to add those collections to SecurityConfig since it would allow for simple tests that look something like if self.security_protocol in SecurityConfig.SASL_MECHANISMS.

self.security_protocol == SecurityConfig.SASL_SSL or \
self.security_protocol == SecurityConfig.SASL_MECHANISM_GSSAPI or \
self.security_protocol == SecurityConfig.SASL_MECHANISM_PLAIN:
cmd += " --sasl-kerberos-service-name %s" % str('kafka')
cmd += " --client-jaas-conf-path %s" % str(SecurityConfig.JAAS_CONF_PATH)
cmd += " --kerb5-conf-path %s" % str(SecurityConfig.KRB5CONF_PATH)

cmd += " 2>> /mnt/kafka_log4j_appender.log | tee -a /mnt/kafka_log4j_appender.log &"
return cmd
Expand Down
17 changes: 12 additions & 5 deletions tests/kafkatest/tests/log4j_appender_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ def __init__(self, test_context):
super(Log4jAppenderTest, self).__init__(test_context)
self.num_zk = 1
self.num_brokers = 1
self.messages_received_count = 0
self.topics = {
TOPIC: {'partitions': 1, 'replication-factor': 1}
}
Expand All @@ -56,13 +57,20 @@ def start_appender(self, security_protocol):
security_protocol=security_protocol)
self.appender.start()

def custom_message_validator(self, msg):
if msg and "INFO : org.apache.kafka.tools.VerifiableLog4jAppender" in msg:
self.logger.debug("Received message: %s" % msg)
self.messages_received_count += 1


def start_consumer(self, security_protocol):
enable_new_consumer = security_protocol == SecurityConfig.SSL
enable_new_consumer = security_protocol != SecurityConfig.PLAINTEXT
self.consumer = ConsoleConsumer(self.test_context, num_nodes=self.num_brokers, kafka=self.kafka, topic=TOPIC,
consumer_timeout_ms=1000, new_consumer=enable_new_consumer)
consumer_timeout_ms=1000, new_consumer=enable_new_consumer,
message_validator=self.custom_message_validator)
self.consumer.start()

@matrix(security_protocol=['PLAINTEXT', 'SSL'])
@matrix(security_protocol=['PLAINTEXT', 'SSL', 'SASL_PLAINTEXT', 'SASL_SSL'])
def test_log4j_appender(self, security_protocol='PLAINTEXT'):
"""
Tests if KafkaLog4jAppender is producing to Kafka topic
Expand All @@ -79,8 +87,7 @@ def test_log4j_appender(self, security_protocol='PLAINTEXT'):
timeout_sec=10, backoff_sec=.2, err_msg="Consumer was too slow to start")

# Verify consumed messages count
expected_lines_count = MAX_MESSAGES * 2 # two times to account for new lines introduced by log4j
wait_until(lambda: len(self.consumer.messages_consumed[1]) == expected_lines_count, timeout_sec=10,
wait_until(lambda: self.messages_received_count == MAX_MESSAGES, timeout_sec=10,
err_msg="Timed out waiting to consume expected number of messages.")

self.consumer.stop()
36 changes: 34 additions & 2 deletions tools/src/main/java/org/apache/kafka/tools/VerifiableLog4jAppender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import net.sourceforge.argparse4j.inf.ArgumentParser;
import net.sourceforge.argparse4j.inf.ArgumentParserException;
import net.sourceforge.argparse4j.inf.Namespace;
import org.apache.kafka.common.protocol.SecurityProtocol;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

Expand Down Expand Up @@ -96,7 +97,7 @@ private static ArgumentParser argParser() {
.required(false)
.setDefault("PLAINTEXT")
.type(String.class)
.choices("PLAINTEXT", "SSL")
.choices("PLAINTEXT", "SSL", "SASL_PLAINTEXT", "SASL_SSL")
.metavar("SECURITY-PROTOCOL")
.dest("securityProtocol")
.help("Security protocol to be used while communicating with Kafka brokers.");
Expand Down Expand Up @@ -124,6 +125,30 @@ private static ArgumentParser argParser() {
.metavar("CONFIG_FILE")
.help("Log4jAppender config properties file.");

parser.addArgument("--sasl-kerberos-service-name")
.action(store())
.required(false)
.type(String.class)
.metavar("SASL-KERBEROS-SERVICE-NAME")
.dest("saslKerberosServiceName")
.help("Name of sasl kerberos service.");

parser.addArgument("--client-jaas-conf-path")
.action(store())
.required(false)
.type(String.class)
.metavar("CLIENT-JAAS-CONF-PATH")
.dest("clientJaasConfPath")
.help("Path of JAAS config file of Kafka client.");

parser.addArgument("--kerb5-conf-path")
.action(store())
.required(false)
.type(String.class)
.metavar("KERB5-CONF-PATH")
.dest("kerb5ConfPath")
.help("Path of Kerb5 config file.");

return parser;
}

Expand Down Expand Up @@ -171,11 +196,18 @@ public static VerifiableLog4jAppender createFromArgs(String[] args) {
props.setProperty("log4j.appender.KAFKA.RequiredNumAcks", res.getString("acks"));
props.setProperty("log4j.appender.KAFKA.SyncSend", "true");
final String securityProtocol = res.getString("securityProtocol");
if (securityProtocol != null && securityProtocol.equals("SSL")) {
if (securityProtocol != null && !securityProtocol.equals(SecurityProtocol.PLAINTEXT.toString())) {
props.setProperty("log4j.appender.KAFKA.SecurityProtocol", securityProtocol);
}
if (securityProtocol != null && securityProtocol.contains("SSL")) {
props.setProperty("log4j.appender.KAFKA.SslTruststoreLocation", res.getString("sslTruststoreLocation"));
props.setProperty("log4j.appender.KAFKA.SslTruststorePassword", res.getString("sslTruststorePassword"));
}
if (securityProtocol != null && securityProtocol.contains("SASL")) {
props.setProperty("log4j.appender.KAFKA.SaslKerberosServiceName", res.getString("saslKerberosServiceName"));
props.setProperty("log4j.appender.KAFKA.clientJaasConfPath", res.getString("clientJaasConfPath"));
props.setProperty("log4j.appender.KAFKA.kerb5ConfPath", res.getString("kerb5ConfPath"));
}
props.setProperty("log4j.logger.kafka.log4j", "INFO, KAFKA");

if (configFile != null) {
Expand Down