Kafka的SSL属性的动态更新不起作用。

我们可以使用AdminClient动态更改Kafka的SSL属性。下面是一个示例代码，展示了如何使用AdminClient动态更新Kafka的SSL证书：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.AlterConfigOp;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.ConfigEntry.AlterConfigOpType;
import org.apache.kafka.common.config.ConfigResource;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;

public class KafkaSSLDynamicUpdate {

  public static void main(String[] args) throws Exception {

    // Kafka broker url
    String bootstrapServers = "<bootstrap_servers>";

    // SSL client truststore password
    String truststorePassword = "<truststore_password>";

    // SSL certificate subject name for Kafka brokers
    String sslCertificateSubjectName = "<certificate_subject_name>";

    // Kafka SSL client properties
    Properties sslProperties = new Properties();
    sslProperties.put("ssl.truststore.location", "<truststore_path>");
    sslProperties.put("ssl.truststore.password", truststorePassword);
    sslProperties.put("ssl.endpoint.identification.algorithm", "");
    sslProperties.put("ssl.key.password", "<key_password>");

    // AdminClient properties
    Properties adminClientProperties = new Properties();
    adminClientProperties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    adminClientProperties.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SSL");
    adminClientProperties.put(AdminClientConfig.SSL_TRUSTSTORE_LOCATION_CONFIG, "<truststore_path>");
    adminClientProperties.put(AdminClientConfig.SSL_TRUSTSTORE_PASSWORD_CONFIG, truststorePassword);
    adminClientProperties.put(AdminClientConfig.SSL_KEY_PASSWORD_CONFIG, "<key_password>");
    adminClientProperties.put(AdminClientConfig.SSL_KEYSTORE_TYPE_CONFIG, "PKCS12");
    adminClientProperties.put(AdminClientConfig.SSL_KEYSTORE_LOCATION_CONFIG, "<keystore_path>");
    adminClientProperties.put(AdminClientConfig.SSL_KEYSTORE_PASSWORD_CONFIG, "<keystore_password>");
    adminClientProperties.put(AdminClientConfig.SSL_KEYSTORE_KEY_PASSWORD_CONFIG, "<key_password>");

    // Get AdminClient instance
    AdminClient admin