public static void initSSL(Properties consumerProps) {
    // Check if one-way SSL is enabled. In this scenario, the client validates the server certificate.
    String trustStoreLocation = consumerProps.getProperty(SSL_TRUSTSTORE_LOCATION);
    String trustStorePassword = consumerProps.getProperty(SSL_TRUSTSTORE_PASSWORD);
    String serverCertificate = consumerProps.getProperty(STREAM_KAFKA_SSL_SERVER_CERTIFICATE);
    if (StringUtils.isAnyEmpty(trustStoreLocation, trustStorePassword, serverCertificate)) {
      LOGGER.info("Skipping auto SSL server validation since it's not configured.");
      return;
    }
    if (shouldRenewTrustStore(consumerProps)) {
      initTrustStore(consumerProps);
    }

    // Set the security protocol
    String securityProtocol = consumerProps.getProperty(SECURITY_PROTOCOL, DEFAULT_SECURITY_PROTOCOL);
    consumerProps.setProperty(SECURITY_PROTOCOL, securityProtocol);

    // Check if two-way SSL is enabled. In this scenario, the client validates the server's certificate and the server
    // validates the client's certificate.
    String keyStoreLocation = consumerProps.getProperty(SSL_KEYSTORE_LOCATION);
    String keyStorePassword = consumerProps.getProperty(SSL_KEYSTORE_PASSWORD);
    String keyPassword = consumerProps.getProperty(SSL_KEY_PASSWORD);
    String clientCertificate = consumerProps.getProperty(STREAM_KAFKA_SSL_CLIENT_CERTIFICATE);
    if (StringUtils.isAnyEmpty(keyStoreLocation, keyStorePassword, keyPassword, clientCertificate)) {
      LOGGER.info("Skipping auto SSL client validation since it's not configured.");
      return;
    }
    if (shouldRenewKeyStore(consumerProps)) {
      initKeyStore(consumerProps);
    }
  }
@Test
  public void testInitSSLBackwardsCompatibilityCheck()
      throws CertificateException, NoSuchAlgorithmException, OperatorCreationException, NoSuchProviderException,
             IOException, KeyStoreException {
    Properties consumerProps = new Properties();
    setTrustStoreProps(consumerProps);
    setKeyStoreProps(consumerProps);
    KafkaSSLUtils.initSSL(consumerProps);

    // validate
    validateTrustStoreCertificateCount(1);
    validateKeyStoreCertificateCount(1);

    setTrustStoreProps(consumerProps); // new server certificate is generated
    consumerProps.remove("stream.kafka.ssl.server.certificate");
    setKeyStoreProps(consumerProps); // new client certificate is generated
    consumerProps.remove("stream.kafka.ssl.client.certificate");

    // Attempt to initialize the trust store and key store again without passing the required certificates
    KafkaSSLUtils.initSSL(consumerProps);
    // validate again that the existing certificates are untouched.
    validateTrustStoreCertificateCount(1);
    validateKeyStoreCertificateCount(1);
  }