@VisibleForTesting
  static void initKeyStore(Properties consumerProps) {
    Path keyStorePath = getKeyStorePath(consumerProps);
    if (Files.exists(keyStorePath)) {
      deleteFile(keyStorePath);
    }
    LOGGER.info("Initializing the SSL key store");
    try {
      // Create the key store path
      createFile(keyStorePath);
    } catch (FileAlreadyExistsException fex) {
      LOGGER.warn("SSL key store initialization failed as key store already exists.");
      return;
    } catch (IOException iex) {
      throw new RuntimeException(String.format("Failed to create the key store path: %s", keyStorePath), iex);
    }

    String keyStorePassword = consumerProps.getProperty(SSL_KEYSTORE_PASSWORD);
    String keyPassword = consumerProps.getProperty(SSL_KEY_PASSWORD);
    String clientCertificate = consumerProps.getProperty(STREAM_KAFKA_SSL_CLIENT_CERTIFICATE);
    String certificateType = consumerProps.getProperty(STREAM_KAFKA_SSL_CERTIFICATE_TYPE, DEFAULT_CERTIFICATE_TYPE);
    String privateKeyString = consumerProps.getProperty(STREAM_KAFKA_SSL_CLIENT_KEY);
    String privateKeyAlgorithm = consumerProps.getProperty(STREAM_KAFKA_SSL_CLIENT_KEY_ALGORITHM,
        DEFAULT_KEY_ALGORITHM);
    String keyStoreType = consumerProps.getProperty(SSL_KEYSTORE_TYPE, DEFAULT_KEYSTORE_TYPE);
    consumerProps.setProperty(SSL_KEYSTORE_TYPE, keyStoreType);

    try {
      // decode the private key and certificate into bytes
      byte[] pkBytes = Base64.getDecoder().decode(privateKeyString);
      byte[] certBytes = Base64.getDecoder().decode(clientCertificate);

      // Create the private key object
      PKCS8EncodedKeySpec keySpec = new PKCS8EncodedKeySpec(pkBytes);
      KeyFactory keyFactory = KeyFactory.getInstance(privateKeyAlgorithm);
      PrivateKey privateKey = keyFactory.generatePrivate(keySpec);

      // Create the Certificate object
      CertificateFactory certFactory = CertificateFactory.getInstance(certificateType);
      InputStream certInputStream = new ByteArrayInputStream(certBytes);
      Certificate certificate = certFactory.generateCertificate(certInputStream);

      // Create a KeyStore object and load a new empty keystore
      KeyStore keyStore = KeyStore.getInstance(keyStoreType);
      keyStore.load(null, null);

      // Add the key pair and certificate to the keystore
      KeyStore.PrivateKeyEntry privateKeyEntry = new KeyStore.PrivateKeyEntry(
          privateKey, new Certificate[]{certificate}
      );
      KeyStore.PasswordProtection keyPasswordProtection = new KeyStore.PasswordProtection(keyPassword.toCharArray());
      keyStore.setEntry(DEFAULT_CLIENT_ALIAS, privateKeyEntry, keyPasswordProtection);

      // Save the keystore to the specified location
      try (FileOutputStream fos = new FileOutputStream(keyStorePath.toString())) {
        keyStore.store(fos, keyStorePassword.toCharArray());
      }
      LOGGER.info("Initialized the SSL key store.");
    } catch (Exception ex) {
      throw new RuntimeException("Error initializing the SSL key store", ex);
    }
  }
@Test
  public void testInitKeyStore()
      throws CertificateException, NoSuchAlgorithmException, OperatorCreationException, NoSuchProviderException,
             IOException, KeyStoreException {
    Properties consumerProps = new Properties();
    setKeyStoreProps(consumerProps);

    // should not throw any exceptions
    KafkaSSLUtils.initKeyStore(consumerProps);
    validateKeyStoreCertificateCount(1);
  }