From 552b7c988e2d4255aa77bd17913c5a70bfd766a3 Mon Sep 17 00:00:00 2001
From: Manikumar Reddy <manikumar.reddy@gmail.com>
Date: Sat, 19 Aug 2017 22:39:48 +0530
Subject: [PATCH] KAFKA-5750: Elevate log messages for denials to INFO level in
 SimpleAclAuthorizer class

---
 config/log4j.properties                                  | 4 ++--
 .../scala/kafka/security/auth/SimpleAclAuthorizer.scala  | 9 +++++++--
 docs/upgrade.html                                        | 1 +
 3 files changed, 10 insertions(+), 4 deletions(-)

diff --git a/config/log4j.properties b/config/log4j.properties
index 0a1ec4f6c1d05..fa5acc90722cf 100644
--- a/config/log4j.properties
+++ b/config/log4j.properties
@@ -86,7 +86,7 @@ log4j.additivity.kafka.log.LogCleaner=false
 log4j.logger.state.change.logger=TRACE, stateChangeAppender
 log4j.additivity.state.change.logger=false
 
-# Change to DEBUG to enable audit log for the authorizer
-log4j.logger.kafka.authorizer.logger=WARN, authorizerAppender
+# Access denials are logged at INFO level. Change to DEBUG to enable all the audit logs for the authorizer
+log4j.logger.kafka.authorizer.logger=INFO, authorizerAppender
 log4j.additivity.kafka.authorizer.logger=false
 
diff --git a/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala b/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
index 6218a2c4b2fd3..1551af60f4f9e 100644
--- a/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
+++ b/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala
@@ -246,8 +246,13 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
   }
 
   private def logAuditMessage(principal: KafkaPrincipal, authorized: Boolean, operation: Operation, resource: Resource, host: String) {
-    val permissionType = if (authorized) "Allowed" else "Denied"
-    authorizerLogger.debug(s"Principal = $principal is $permissionType Operation = $operation from host = $host on resource = $resource")
+    def logMessage: String = {
+      val result = if (authorized) "Allowed" else "Denied"
+      s"Principal = $principal is $result Operation = $operation from host = $host on resource = $resource"
+    }
+
+    if (authorized) authorizerLogger.debug(logMessage)
+    else authorizerLogger.info(logMessage)
   }
 
   /**
diff --git a/docs/upgrade.html b/docs/upgrade.html
index d265bc937b25f..22d89cdeb21cd 100644
--- a/docs/upgrade.html
+++ b/docs/upgrade.html
@@ -67,6 +67,7 @@ <h5><a id="upgrade_100_notable" href="#upgrade_100_notable">Notable changes in 1
         A stub implementation has been retained for binary compatibility.</li>
     <li>The Java clients and tools now accept any string as a client-id.</li>
     <li>The deprecated tool <code>kafka-consumer-offset-checker.sh</code> has been removed. Use <code>kafka-consumer-groups.sh</code> to get consumer group details.</li>
+    <li>In SimpleAclAuthorizer class, access denials are logged at INFO level and access approvals are logged at DEBUG level.</li>
 </ul>
 
 <h5><a id="upgrade_100_new_protocols" href="#upgrade_100_new_protocols">New Protocol Versions</a></h5>