ApsaraMQ for Kafka:Use instance endpoints to send and receive messages

Compile and run KafkaProducerDemo.java to send messages.

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
// If you use an SSL or SASL endpoint, comment out the following line.
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
/*
* If you use an SSL or SASL endpoint, uncomment the following two lines.
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/

public class KafkaProducerDemo {

    public static void main(String args[]) {
          
       /*
        * If you use an SSL endpoint, uncomment the following line.
        * Set the path of the JAAS configuration file.
        JavaKafkaConfigurer.configureSasl();
        */
         
       /*
        * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following line.
        * Set the path of the JAAS configuration file.
        JavaKafkaConfigurer.configureSaslPlain();
        */
       
       /*
        * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following line.
        * Set the path of the JAAS configuration file.
        JavaKafkaConfigurer.configureSaslScram();
        */

        // Load kafka.properties.
        Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        // Set the endpoint. Obtain the endpoint for the topic from the console.
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
         
       /*
        * If you use an SSL endpoint, uncomment the following four lines.
        * Similar to the SASL path, this file cannot be packaged into the JAR file.
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
        * The password for the root certificate store. Keep the value unchanged.
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
        * The access protocol. SASL_SSL is supported.
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        * The SASL authentication method. Keep the value unchanged.
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        */

       /*
        * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following two lines.
        * Access protocol.
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * PLAIN mechanism.
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        */

       /*
        * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following two lines.
        * Access protocol.
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * SCRAM mechanism.
        props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
        */

        // The serialization method for ApsaraMQ 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 wait time for a request.
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
        // The number of retries on the client.
        props.put(ProducerConfig.RETRIES_CONFIG, 5);
        // The retry interval on the client.
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3000);
         
       /*
        * If you use an SSL endpoint, uncomment the following line.
        * Set hostname verification to empty.
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
        */

        // Construct a producer object. This object is thread-safe. Typically, one producer object is sufficient for a process.
        // To improve performance, you can create more producer objects, but do not create too many. The number of producer objects should not exceed five.
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        // Construct an ApsaraMQ for Kafka message.
        String topic = kafkaProperties.getProperty("topic"); // The topic to which the message belongs. After you create the topic in the console, enter the topic name here.
        String value = "this is the message's value"; // The content of the message.

        try {
            // Obtaining future objects in batches can improve performance. Do not use excessively large batches.
            List<Future<RecordMetadata>> futures = new ArrayList<Future<RecordMetadata>>(128);
            for (int i =0; i < 100; i++) {
                // Send the message and obtain a future object.
                ProducerRecord<String, String> kafkaMessage =  new ProducerRecord<String, String>(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.
                try {
                    RecordMetadata recordMetadata = future.get();
                    System.out.println("Produce ok:" + recordMetadata.toString());
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        } catch (Exception e) {
            // If the message fails to be sent after several retries on the client, your application must handle the error.
            System.out.println("error occurred");
            e.printStackTrace();
        }
    }
}