KAFKA-3396 : Unauthorized topics are returned to the user

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

@@ -670,6 +670,10 @@ public void handle(OffsetCommitResponse commitResponse, RequestFuture<Void> futu
                     resetGeneration();
                     future.raise(new CommitFailedException());
                     return;
+                } else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+                    log.debug("Offset commit for group {} failed on partition {}: {}", groupId, tp, error.message());
+                    future.raise(new KafkaException("Partition " + tp + " may not exist or user may not have Describe access to topic: " + error.message()));
+                    return;
                 } else {
                     log.error("Group {} failed to commit partition {} at offset {}: {}", groupId, tp, offset, error.message());
                     future.raise(new KafkaException("Unexpected error in commit: " + error.message()));
@@ -731,6 +735,8 @@ public void handle(OffsetFetchResponse response, RequestFuture<Map<TopicPartitio
                         // re-discover the coordinator and retry
                         coordinatorDead();
                         future.raise(error);
+                    } else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+                        future.raise(new KafkaException("Partition " + tp + " may not exist or user may not have Describe access to topic: " + error.message()));
                     } else {
                         future.raise(new KafkaException("Unexpected error in fetch offset response: " + error.message()));
                     }

core/src/main/scala/kafka/admin/AdminUtils.scala

@@ -26,7 +26,7 @@ import kafka.utils.ZkUtils._
 import java.util.Random
 import java.util.Properties
 import org.apache.kafka.common.Node
-import org.apache.kafka.common.errors.{ReplicaNotAvailableException, InvalidTopicException, LeaderNotAvailableException, InvalidPartitionsException, InvalidReplicationFactorException, TopicExistsException, InvalidReplicaAssignmentException}
+import org.apache.kafka.common.errors.{ReplicaNotAvailableException, UnknownTopicOrPartitionException, InvalidTopicException, LeaderNotAvailableException, InvalidPartitionsException, InvalidReplicationFactorException, TopicExistsException, InvalidReplicaAssignmentException}
 import org.apache.kafka.common.protocol.{Errors, SecurityProtocol}
 import org.apache.kafka.common.requests.MetadataResponse
 
@@ -325,7 +325,7 @@ object AdminUtils extends Logging with AdminUtilities {
           case e2: Throwable => throw new AdminOperationException(e2)
         }
       } else {
-        throw new InvalidTopicException("topic %s to delete does not exist".format(topic))
+        throw new UnknownTopicOrPartitionException("topic %s to delete does not exist".format(topic))
       }
     }
 

core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala

@@ -127,9 +127,9 @@ class SimpleAclAuthorizer extends Authorizer with Logging {
     //check if there is any Deny acl match that would disallow this operation.
     val denyMatch = aclMatch(session, operation, resource, principal, host, Deny, acls)
 
-    //if principal is allowed to read or write we allow describe by default, the reverse does not apply to Deny.
+    //if principal is allowed to read, write or delete we allow describe by default, the reverse does not apply to Deny.
     val ops = if (Describe == operation)
-      Set[Operation](operation, Read, Write)
+      Set[Operation](operation, Read, Write, Delete)
     else
       Set[Operation](operation)
 

core/src/main/scala/kafka/server/MetadataCache.scala

@@ -120,12 +120,6 @@ private[server] class MetadataCache(brokerId: Int) extends Logging {
     }
   }
 
-  def hasTopicMetadata(topic: String): Boolean = {
-    inReadLock(partitionMetadataLock) {
-      cache.contains(topic)
-    }
-  }
-
   def getAllTopics(): Set[String] = {
     inReadLock(partitionMetadataLock) {
       cache.keySet.toSet