title: Why Kafka Streams require DESCRIBE access to other topics when kafka cluster is configured with SASL_SSL?
body: <p>I have a 3 Node Kafka cluster (Kafka Version 1.1.1) with SASL_SSL security enabled using pluggable JAAS configurations. When Kafka Producers and consumers with proper security configurations communicate with secured broker then write/read operations works without any issue, when I use Kafka streams then I am seeing a weird behaviour even if the Stream application is working fine on client side, I see an error log in the server side where threads corresponding to stream application are making a describe request to <strong>all other topics</strong> on the broker. My Stream application is very simple which copies data from a topic "testA" to another topic "testB". Why are Kafka stream threads on broker side trying to describe other topics which steam application is not even authorised to describe. Ideally the stream application should only describe the topics "testA" and "testB", if my understanding is correct. </p>
  
  <p>The error log is :</p>
  
  <blockquote>
    <p>[2020-01-16 09:28:02,235] ERROR Requested Operation : DESCRIBE for
    Resource : Topic:<strong>dmp-tgd8uye1am</strong> cannot be authorized due to
    invalid/missing claims in the JWT_TOKEN claims :  {
        Topic:<strong>testA</strong>=[READ, WRITE, DESCRIBE], 
        Topic:<strong>testB</strong>=[READ, WRITE, DESCRIBE], 
        Group:*=[READ, WRITE, DESCRIBE] } (com.core.security.providers.kafka.authorization.DmpJwtAuthorizer)</p>
  </blockquote>
  
  <p>The JWT_TOKEN is set by the Client side with permissions for "<strong>testA</strong>" and "<strong>testB</strong>". Why in the above error log stream thread is making describe call to "<strong>dmp-tgd8uye1am</strong>" topic ?</p>
  
  <p>This token is validated on the server side involving 2 steps a) Authentication b) Authorisation</p>
  
  <p>a) <strong>Authentication with SaslServer class</strong> : This step is successful.</p>
  
  <p>b) <strong>Authorisation with DmpJwtAuthorizer class (<em>implementation details below</em>)</strong> : This is the step where the weird logs show up. I can see that in the broker process logs there are multiple threads corresponding to stream application doing authorisation. The threads which request describe access for "testA" and "testB" passes. And the remaining threads request for DESCRIBE access on other topics like  "<strong>dmp-tgd8uye1am</strong>" which stream application has nothing to do with. </p>
  
  <p>The broker security configurations are as follows :</p>
  
  <ol>
  <li>Kafka-server-start.sh</li>
  </ol>
  
  <pre><code>export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka_2.11-1.1.1/kafka-config/kafka_server_jaas.conf"
  </code></pre>
  
  <ol start="2">
  <li>kafka_server_jaas.conf</li>
  </ol>
  
  <pre><code>KafkaServer {
     com.core.security.providers.kafka.authentication.DmpJwtLoginModule required
     username="kafkabroker"
     password="*********"
     user_kafkabroker="**********";
  };
  Client {
     org.apache.zookeeper.server.auth.DigestLoginModule required
     username="kafka"
     password="**********";
  };
  </code></pre>
  
  <ol start="3">
  <li>Server.properties</li>
  </ol>
  
  <pre><code>listeners=SSL://:9093,SASL_SSL://:9094
  advertised.listeners=SSL://:9093,SASL_SSL://:9094
  sasl.enabled.mechanisms=PLAIN
  authorizer.class.name=com.core.security.providers.kafka.authorization.DmpJwtAuthorizer
  principal.builder.class=com.core.security.providers.kafka.authentication.DmpJwtPrincipalBuilder
  security.inter.broker.protocol=SSL
  ssl.client.auth=required
  ssl.truststore.location=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.222.b10-0.amzn2.0.1.x86_64/jre/lib/security/cacerts
  ssl.truststore.password=******
  ssl.keystore.location=/opt/kafka_2.11-1.1.1/kafka-config/keystore
  ssl.keystore.password=*****
  
  default.replication.factor=3
  min.insync.replicas=2
  auto.leader.rebalance.enable=true
  </code></pre>
  
  <p>The KafkaStream client configs and working logic are as follows :</p>
  
  <pre><code>configStreamProperties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().configStreamProperties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
  configStreamProperties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * configStreamProperties.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, configStreamProperties.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, configStreamProperties.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, "2");
  configProperties.put("security.protocol", "SASL_SSL");
  configProperties.put("sasl.mechanism", "PLAIN");
  configProperties.put("sasl.jaas.config", "com.core.security.providers.kafka.authentication.DmpJwtClient required;");
  
  
  final StreamsBuilder builder = new StreamsBuilder();
  
  String inputTopic = (String) inputOptions.get("testA");
  String outputTopic = (String) outputOptions.get("testB");
  
  builder.stream(inputTopic).to(outputTopic);
  
  final KafkaStreams streams = new KafkaStreams(builder.build(), configStreamProperties);
  
  streams.cleanUp();
  streams.start();    
  
  Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
  </code></pre>
  
  <p>DmpJwtAuthorizer Class on Server Side :</p>
  
  <pre><code>public class DmpJwtAuthorizer extends kafka.security.auth.SimpleAclAuthorizer {
  
  // SimpleAclAuthorizer is part of Kafka libraries and extends the kafka.security.auth.Authorizer class
  
      private static final Logger logger = LoggerFactory.getLogger(DmpJwtAuthorizer.class);
  
      @Override
      public boolean authorize(Session session, Operation operation, Resource resource) {
          boolean authorized = false;
  
          DmpKafkaPrincipal principal = (DmpKafkaPrincipal) session.principal();
          String user = principal.getName();
          Map&lt;Resource, List&lt;AclOperation&gt;&gt; allowedResources = principal.getAllowedResources();
  
          if ((user != null) &amp;&amp; (!user.isEmpty())) {
              Iterator&lt;Entry&lt;Resource, List&lt;AclOperation&gt;&gt;&gt; iterator = allowedResources.entrySet().iterator();
  
              while(iterator.hasNext() &amp;&amp; !authorized) {
                  Entry&lt;Resource, List&lt;AclOperation&gt;&gt; entry = iterator.next();
                  Resource allowedResource = entry.getKey();
                  List&lt;AclOperation&gt; allowedOperations = entry.getValue();
  
                  AclOperation requestedACLOperation = operation.toJava();
  
                  if(resource.resourceType().name().equals(allowedResource.resourceType().name())
                          &amp;&amp; isMatchingResourceName(resource, allowedResource)
                          &amp;&amp; allowedOperations.contains(requestedACLOperation)){
  
                      authorized = true;
                  }
              }
  
              if (!authorized) {
  
                logger.error("User : {} Requested Operation : {} for Resource : {} cannot be authorized due to invalid/missing claims in the JWT_TOKEN claims : {} THREAD NAME : {} THREAD ID : {}", user, operation.toJava(), resource, allowedResources, Thread.currentThread().getName(), Thread.currentThread().getId());
              }
          }
  
          return authorized;
      }
  
      public boolean isMatchingResourceName(Resource requestedResource, Resource allowedResource){
          String requestedResourceName = requestedResource.name();
          String allowedResourceName = allowedResource.name();
  
          String regex = ("\\Q" + allowedResourceName + "\\E").replace("*", "\\E.*\\Q");
          return requestedResourceName.matches(regex);
      }
  }
  </code></pre>