ApsaraMQ for Kafka:Send messages from Logstash to ApsaraMQ for Kafka over the Internet

Download the SSL certificate

Switch to the Logstash bin directory and download the truststore certificate:

cd <logstash-install-dir>/bin
wget -O kafka.client.truststore.jks https://github.com/AliwareMQ/aliware-kafka-demos/raw/master/kafka-log-stash-demo/vpc-ssl/mix.4096.client.truststore.jks

Create the JAAS configuration file

Create a file named jaas.conf in the Logstash bin directory with the following content:

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="<your-username>"
  password="<your-password>";
};

Replace the placeholders with your actual values:

Create the Logstash output configuration file

Create a file named output.conf in the Logstash bin directory with the following content:

input {
    stdin {}
}

output {
    stdout { codec => json }
    kafka {
        bootstrap_servers => "<your-endpoint>"
        topic_id => "<your-topic>"
        security_protocol => "SASL_SSL"
        sasl_mechanism => "PLAIN"
        jaas_path => "<logstash-install-dir>/bin/jaas.conf"
        ssl_truststore_password => "KafkaOnsClient"
        ssl_truststore_location => "<logstash-install-dir>/bin/kafka.client.truststore.jks"
        ssl_endpoint_identification_algorithm => ""
    }
}

Replace these placeholders with your actual values:

The remaining parameters use fixed values. Do not change them:

Send a test message

1. Start Logstash with the output configuration:

      ./logstash -f output.conf

2. After Logstash starts, type test and press Enter. The stdout output displays the message locally as JSON, while the kafka output sends it to your ApsaraMQ for Kafka instance.

   

Check partition status

1. Log on to the ApsaraMQ for Kafka console.

2. In the Resource Distribution section of the Overview page, select the region where your instance resides.

3. On the Instances page, click the name of the target instance.

4. In the left-side navigation pane, click Topics.

5. Click the topic name, then click the Partition Status tab on the Topic Details page.

   Parameter	Description
   Partition ID	The partition ID.
   Minimum Offset	The earliest offset in the partition.
   Maximum Offset	The latest offset in the partition.
   Messages	The total number of messages in the partition.
   Last Updated At	The time when the most recent message was stored.

   

Query a message by offset

1. Log on to the ApsaraMQ for Kafka console.

2. In the Resource Distribution section of the Overview page, select the region where your instance resides.

3. On the Instances page, click the name of the target instance.

4. In the left-side navigation pane, click Message Query.

5. From the Search Method drop-down list, select Search by offset.

6. Select a topic from the Topic drop-down list, select a partition from the Partition drop-down list, enter an offset value in the Offset field, and click Search. The console returns all messages with offsets greater than or equal to the specified value. For example, setting Partition to 5 and Offset to 5 returns all messages from partition 5 with offsets 5 and above.

   Parameter	Description
   Partition	The partition where the message is stored.
   Offset	The message offset within the partition.
   Key	The message key, displayed as a string.
   Value	The message body, displayed as a string.
   Created At	The timestamp when the message was sent. If you specified a value for the ProducerRecord timestamp field, the specified value is displayed. If you did not specify a value, the local system time when the message was sent is displayed. If the timestamp field is set to 0 or an invalid value, the time is displayed in 1970/x/x x:x:x format. Clients on ApsaraMQ for Kafka version 0.9 or earlier cannot set this field.
   Actions	Download Key: Download the message key. Download Value: Download the message body. The console displays up to 1 KB per message. Download the message to view content beyond 1 KB. A maximum of 10 MB of messages can be downloaded at a time.