# Truststore path and password, used to verify the server certificate
ssl.truststore.location=/Users/ckafka/client.truststore.jks # jks file path
ssl.truststore.password=your_password
# SASL account and password configuration
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \\
   username="yourinstance#yourusername" \\
   password="your_password"; 
ssl.endpoint.identification.algorithm=

# 1. Verify the server identity 
ssl.truststore.location=/Users/ckafka/client.truststore.jks  # jks file path
ssl.truststore.password=your_password

# SASL account and password configuration
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \\
   username="your_username" \\
   password="your_password";

# 2. Provide the client identity 
ssl.keystore.location=/data/ssl/client.keystore.jks
ssl.keystore.password=client_password  # The password you set when generating the certificate
ssl.key.password=key_password           # Private key password

# Note: For two-way authentication, ensure that ssl.endpoint.identification.algorithm is empty or set to "".
# Avoid handshake failures caused by domain name validation
ssl.endpoint.identification.algorithm=

# Truststore path and password, used to verify the server certificate
ssl.truststore.location=/Users/ckafka/client.truststore.jks # jks file path
ssl.truststore.password=5fi6R!M
# SASL account and password configuration
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \\
   username="yourinstance#yourusername" \\
   password="your_password"; 
ssl.endpoint.identification.algorithm=

public class CKafkaConfigurer {
    private static Properties properties;
    public static void configureSaslPlain() {
        //If it has been set using -D or other methods, it will not be set here.
        if (null == System.getProperty("java.security.auth.login.config")) {
            //Be sure to change XXX to your own path.
            System.setProperty("java.security.auth.login.config",
                    getCKafkaProperties().getProperty("java.security.auth.login.config.plain"));
        }
    }
    public synchronized static Properties getCKafkaProperties() {
        if (null != properties) {
            return properties;
        }
        //Obtain the content of the configuration file kafka.properties.
        Properties kafkaProperties = new Properties();
        try {
            kafkaProperties.load(CKafkaProducerDemo.class.getClassLoader().getResourceAsStream("kafka.properties"));
        } catch (Exception e) {
            System.out.println("getCKafkaProperties error");
        }
        properties = kafkaProperties;
        return kafkaProperties;
    }
}

public class KafkaSaslProducerDemo {
   public static void main(String[] args) {
      //Set the path to the JAAS configuration file.
      CKafkaConfigurer.configureSaslPlain();
      //Load kafka.properties.
      Properties kafkaProperties = CKafkaConfigurer.getCKafkaProperties();
      Properties props = new Properties();
      //Set the access point. You can obtain the access point of the corresponding topic in the console.
      props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
              kafkaProperties.getProperty("bootstrap.servers"));
      //
      //  SASL_SSL access in the public network.
      //
      //  Access protocol.
      props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
      //  The plaintext method of SASL authentication is used.
      props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
      //  SSL encryption.
      props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG));
      props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, kafkaProperties.getProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG));
      props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,kafkaProperties.getProperty(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG));
      //Serialization method for Kafka messages.
      props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
              "org.apache.kafka.common.serialization.StringSerializer");
      props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
              "org.apache.kafka.common.serialization.StringSerializer");
      //The maximum request waiting time.
      props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
      //Set the number of internal retries on the client.
      props.put(ProducerConfig.RETRIES_CONFIG, 5);
      //Set the interval of internal retries on the client.
      props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
      //ack=0:   The producer will not wait for acknowledgment from the broker, and the retry configuration will not take effect. Note that the connection will be closed if traffic is throttled.
      //ack=1:   The broker leader will return an acknowledgment without waiting for acknowledgments from all broker followers.
      //ack=all: The broker leader will return an acknowledgment only after receiving acknowledgments from all broker followers.
      props.put(ProducerConfig.ACKS_CONFIG, "all");
      //Construct a producer object. Note that this object is thread-safe. Generally, only one producer object is required in one process.
      KafkaProducer<String, String> producer = new KafkaProducer<>(props);
      //Construct a Kafka message.
      String topic = kafkaProperties.getProperty("topic"); //The topic to which the message belongs. Fill it in here after applying in the console.
      String value = "this is ckafka msg value"; //The content of the message.
      try {
         //Obtaining Future objects in batches can speed up the process. Be careful not to use too large a batch.
         List<Future<RecordMetadata>> futures = new ArrayList<>(128);
         for (int i = 0; i < 100; i++) {
            //Send a message and obtain a Future object.
            ProducerRecord<String, String> kafkaMessage = new ProducerRecord<>(topic,
                    value + ": " + i);
            Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
            futures.add(metadataFuture);
         }
         producer.flush();
         for (Future<RecordMetadata> future : futures) {
            //Synchronously obtain the result of the Future object.
            RecordMetadata recordMetadata = future.get();
            System.out.println("Produce ok:" + recordMetadata.toString());
         }
      } catch (Exception e) {
         //After the client retries internally, the sending still fails. The service needs to respond to this type of error.
         System.out.println("error occurred");
      }
   }
}

public class KafkaSaslConsumerDemo {
   public static void main(String[] args) {
      //Set the path to the JAAS configuration file.
      CKafkaConfigurer.configureSaslPlain();
      //Load kafka.properties.
      Properties kafkaProperties = CKafkaConfigurer.getCKafkaProperties();
      Properties props = new Properties();
      //Set the access point. You can obtain the access point of the corresponding topic in the console.
      props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
              kafkaProperties.getProperty("bootstrap.servers"));
      //
      //  SASL_SSL access in the public network.
      //
      //  Access protocol.
      props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
      //  The plaintext method of SASL authentication is used.
      props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
      //  SSL encryption.
      props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG));
      props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, kafkaProperties.getProperty(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG));
      props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,kafkaProperties.getProperty(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG));
      //Consumer timeout duration.
      //The default value is 30 seconds. If a consumer fails to return a heartbeat within the duration, the server determines that the consumer is not alive, removes it from the consumer group, and triggers the rebalance process.
      props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
      //Maximum interval between two polls.
      //For versions earlier than 0.10.1.0, the 2 concepts are mixed and both represented by session.timeout.ms.
      props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30000);
      //Maximum number per poll.
      //Be careful that this value cannot be too large. If too much data is polled and cannot be consumed before the next poll, a load balancing will be triggered, resulting in delays.
      props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
      //Deserialization method of messages.
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
              "org.apache.kafka.common.serialization.StringDeserializer");
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
              "org.apache.kafka.common.serialization.StringDeserializer");
      //The consumer group to which the current consumption instance belongs. Fill it in after applying in the console.
      //Consumption instances of the same group will carry consumption messages.
      props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
      //Construct a consumption object, that is, generate a consumer instance.
      KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
      //Set the topics subscribed by the consumer group. Multiple topics can be subscribed to.
      //If GROUP_ID_CONFIG is the same, it is recommended to set the subscribed topic as the same.
      List<String> subscribedTopics = new ArrayList<String>();
      //If multiple topics need to be subscribed to, add them here.
      //Create each topic in the console first.
      String topicStr = kafkaProperties.getProperty("topic");
      String[] topics = topicStr.split(",");
      for (String topic : topics) {
         subscribedTopics.add(topic.trim());
      }
      consumer.subscribe(subscribedTopics);
      //Consume messages in loops.
      while (true) {
         try {
            ConsumerRecords<String, String> records = consumer.poll(1000);
            //The data must be consumed before the next poll, and the total time cannot exceed the value of SESSION_TIMEOUT_MS_CONFIG.
            for (ConsumerRecord<String, String> record : records) {
               System.out.println(
                       String.format("Consume partition:%d offset:%d", record.partition(),
                               record.offset()));
            }
         } catch (Exception e) {
            System.out.println("consumer error!");
         }
      }
   }
}