title: Kafka java client based on SASL_SSL, ignoring configs for truststore and taking default certs
body: <p>The consumer I am trying using default certs for handshake, ignoring the certs defined in Config, what is went wrong? </p>
  
  <p>When trying to consume messages, it is using default certs and  raising SSL Handshake error.</p>
  
  <pre><code>    public class Consumer {
  
      final KafkaConsumer&lt;String,String&gt; mConsumer;
      final Logger mLogger = LoggerFactory.getLogger(Consumer.class);
  
      private Properties consumerProperties(String bootstrapServer,String username, String password,String certPasswd) {
          //String serializer = StringSerializer.class.getName();
          String serializer = StringSerializer.class.getName();
          String deserializer = StringDeserializer.class.getName();
          String jaasTemplate = "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";";
          String jaasCfg = String.format(jaasTemplate, username, password);
          Properties props = new Properties();
          props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServer);
          props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, serializer);
          props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,serializer);
          props.setProperty("security.inter.broker.protocol", "SASL_SSL");
          props.setProperty("bootstrap.servers", bootstrapServer);
          props.setProperty("group.id", username + "-nprod-consumer");
          //props.setProperty("enable.auto.commit", "true");
          props.setProperty("auto.commit.interval.ms", "1000");
          props.setProperty("auto.offset.reset", "earliest");
          props.setProperty("session.timeout.ms", "30000");
          props.setProperty("key.deserializer", deserializer);
          props.setProperty("value.deserializer", deserializer);
          props.setProperty("key.serializer", serializer);
          props.setProperty("value.serializer", serializer);
          props.setProperty("security.protocol", "SASL_SSL");
          props.setProperty("sasl.mechanism", "PLAIN");
          props.setProperty("ssl.keystore.location","client.keystore.jks");
          props.setProperty("ssl.keystore.password",certPasswd);
          props.setProperty("ssl.key.password",certPasswd);
          props.setProperty("ssl.enabled.protocols","TLSv1.2,TLSv1.1,TLSv1");
          props.setProperty("ssl.client.auth","required");
          props.setProperty("ssl.truststore.location","client.truststore.jks");
          props.setProperty("ssl.truststore.password",certPasswd);
          props.setProperty("sasl.jaas.config", jaasCfg);
          return props;
      }
  
      Consumer(String bootstrapServer,String username, String password,String certPassword) {
          Properties props = consumerProperties(bootstrapServer,username,password,certPassword);
          mConsumer= new KafkaConsumer&lt;&gt;(props);
          mLogger.info("Consumer Initialized");
      }
  
      void read(String topic) throws ExecutionException, InterruptedException {
          mConsumer.subscribe(Arrays.asList(topic));
          while (true) {
              ConsumerRecords&lt;String, String&gt; records = mConsumer.poll(1000);
              for (ConsumerRecord&lt;String, String&gt; record : records) {
                  System.out.printf("%s [%d] offset=%d, key=%s, value=\"%s\"\n",
                                    record.topic(), record.partition(),
                                    record.offset(), record.key(), record.value());
              }
          }
      }
      void close(){
          mLogger.info("Closing Consumer");
          mConsumer.close();
      }
  }
  </code></pre>
  
  <p>Please advice, how I can ignore default certs and enforce the certs defined in configs for consumer.</p>