KAFKA-5750; Elevate log messages for denials to INFO in SimpleAclAuth…

…orizer Author: Manikumar Reddy <[email protected]> Reviewers: Ismael Juma <[email protected]> Closes apache#3698 from omkreddy/KAFKA-5750
tulusyosef · Sep 27, 2017 · 8dfdcfd · 8dfdcfd
1 parent 5b943ca
commit 8dfdcfd
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 4 deletions.
diff --git 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 also log allowed accesses
+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
@@ -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 authResult = if (authorized) "Allowed" else "Denied"
+      s"Principal = $principal is $authResult 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
@@ -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>SimpleAclAuthorizer now logs access denials to the authorizer log by default.</li>
 </ul>
 
 <h5><a id="upgrade_100_new_protocols" href="#upgrade_100_new_protocols">New Protocol Versions</a></h5>