static void setupKafkaSslProperties(Configuration configuration, Properties props) {
    copySSLProperties(configuration, props);
    // Setup SSL via credentials keystore if necessary
    final String credKeystore = configuration.get(KafkaTableProperties.HIVE_KAFKA_SSL_CREDENTIAL_KEYSTORE.getName());
    if (!(credKeystore == null) && !credKeystore.isEmpty()) {
      final String truststorePasswdConfig =
          configuration.get(KafkaTableProperties.HIVE_KAFKA_SSL_TRUSTSTORE_PASSWORD.getName());
      final String keystorePasswdConfig =
          configuration.get(KafkaTableProperties.HIVE_KAFKA_SSL_KEYSTORE_PASSWORD.getName());
      final String keyPasswdConfig = configuration.get(KafkaTableProperties.HIVE_KAFKA_SSL_KEY_PASSWORD.getName());

      String resourcesDir = HiveConf.getVar(configuration, HiveConf.ConfVars.DOWNLOADED_RESOURCES_DIR);
      try {
        String truststoreLoc = configuration.get(KafkaTableProperties.HIVE_SSL_TRUSTSTORE_LOCATION_CONFIG.getName());
        Path truststorePath = new Path(truststoreLoc);
        props.setProperty(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
            new File(resourcesDir + "/" + truststorePath.getName()).getAbsolutePath());
        writeStoreToLocal(configuration, truststoreLoc, new File(resourcesDir).getAbsolutePath());

        final String truststorePasswd = Utilities.getPasswdFromKeystore(credKeystore, truststorePasswdConfig);
        props.setProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, truststorePasswd);

        // ssl.keystore.password is only needed if two-way authentication is configured.
        if(!keystorePasswdConfig.isEmpty()) {
          log.info("Kafka keystore configured, configuring local keystore");
          String keystoreLoc = configuration.get(KafkaTableProperties.HIVE_SSL_KEYSTORE_LOCATION_CONFIG.getName());
          Path keystorePath = new Path(keystoreLoc);
          props.setProperty(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,
              new File(resourcesDir + "/" + keystorePath.getName()).getAbsolutePath());
          writeStoreToLocal(configuration, keystoreLoc, new File(resourcesDir).getAbsolutePath());

          final String keystorePasswd = Utilities.getPasswdFromKeystore(credKeystore, keystorePasswdConfig);
          props.setProperty(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, keystorePasswd);
        }

        // ssl.key.password is optional for clients.
        if(!keyPasswdConfig.isEmpty()) {
          final String keyPasswd = Utilities.getPasswdFromKeystore(credKeystore, keyPasswdConfig);
          props.setProperty(SslConfigs.SSL_KEY_PASSWORD_CONFIG, keyPasswd);
        }
      } catch (IOException | URISyntaxException e) {
        throw new IllegalStateException("Unable to retrieve password from the credential keystore", e);
      }
    }
  }
@param configuration Job configs

@return default consumer properties
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaUtils.java