title: Kafka producer ERROR using "security.protocol"
body: <p>I have a Kerberized cluster with 2 Kafka brokers and 3 Zookeeper nodes.
  I have a Java application, in local, ( using kafka-client 2.4.0 ) that must produce and read messages from topics in Kafka brokers.
  In input at the VM I gave:</p>
  
  <pre><code>-Djava.security.auth.login.config=/Users/mypath/kafka_client_jaas.conf
  </code></pre>
  
  <p>I also have a schema registry ( confluent ), in local, connected to the cluster.
  To create a Producer a set those options:</p>
  
  <pre><code>    Properties producerProps = new Properties();
      producerProps.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
      producerProps.put("sasl.kerberos.service.name", "kafka");
      producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_BROKERS);
      producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, user);
      producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
      producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
      producerProps.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
  </code></pre>
  
  <p><strong>error:</strong></p>
  
  <pre><code>java.io.IOException: Configuration Error:
      row 5: expected [option key], found [null]
      at sun.security.provider.ConfigFile$Spi.&lt;init&gt;(ConfigFile.java:137)
      at sun.security.provider.ConfigFile.&lt;init&gt;(ConfigFile.java:102)
  </code></pre>
  
  <p><strong>The problem is caused by</strong> </p>
  
  <pre><code>producerProps.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
  </code></pre>
  
  <p>Without this line, the application compiles but I receive:</p>
  
  <pre><code>org.apache.kafka.common.errors.TimeoutException: Topic TEST not present in metadata after 60000 ms.
  </code></pre>
  
  <p>What should I do? </p>
  
  <p>P.S. the "<code>kafka_client_jaas.conf</code>" should be right, since I use it also with schema registry and it works fine.</p>
  
  <p><strong>kafka_client_jaas.conf:</strong></p>
  
  <pre><code>KafkaClient { com.sun.security.auth.module.Krb5LoginModule required 
  useKeyTab=true 
  storeKey=true 
  keyTab="/Users/path/kafka.service.keytab" \ 
  principal="kafka/kafka-broker01.domain.xx@DOMAIN.XX"; };
  </code></pre>