ApsaraMQ for Kafka:Consume messages from ApsaraMQ for Kafka with Logstash over the internet

Create a topic

1. Log in to the ApsaraMQ for Kafka console.

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

   Important

   Create the topic in the same region as the Elastic Compute Service (ECS) instance that runs your producers and consumers. Topics cannot be used across regions.

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

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

5. Click Create Topic.

6. In the Create Topic panel, configure the following parameters and click OK.

   Parameter	Description	Example
   Name	The topic name.	demo
   Description	The topic description.	demo test
   Partitions	The number of partitions.	12
   Storage Engine	The storage engine type. Cloud Storage uses Alibaba Cloud disks with three replicas in distributed mode, offering low latency and high reliability. Local Storage uses the in-sync replicas (ISR) algorithm of open source Apache Kafka. The storage engine is only configurable on non-serverless Professional Edition instances. Other instance types default to Cloud Storage. Standard (High Write) edition instances support only Cloud Storage.	Cloud Storage
   Message Type	Normal Message: Messages with the same key are stored in the same partition in send order. Ordering may not be preserved if a broker fails. Automatically set when you select Cloud Storage. Partitionally Ordered Message: Same ordering behavior, but ordering is preserved even during broker failures (some partitions may be temporarily unavailable). Automatically set when you select Local Storage.	Normal Message
   Log Cleanup Policy	Only available with Local Storage on Professional Edition instances. Delete (default): Retains messages based on the maximum retention period. Deletes the earliest messages when storage exceeds 85%. Compact: Retains the latest value for each message key. Required by components such as Kafka Connect and Confluent Schema Registry. For more information, see aliware-kafka-demos.	Compact
   Tag	Optional tags for the topic.	demo

Create a consumer group

1. On the instance details page, click Groups in the left-side navigation pane.

2. Click Create Group.

3. In the Create Group panel, enter a name in the Group ID field, add an optional description, and click OK.

Send test messages

To verify the connection later, send a test message to the topic:

1. On the instance details page, click Topics in the left-side navigation pane.

2. Click the topic name, then click Send Message in the upper-right corner of the Topic Details page.

3. In the Start to Send and Consume Message panel, select a Sending Method (Console, Docker, or SDK) and follow the on-screen instructions to send a test message.

Download the SSL certificate

wget -O kafka.client.truststore.jks https://github.com/AliwareMQ/aliware-kafka-demos/blob/master/kafka-log-stash-demo/vpc-ssl/mix.4096.client.truststore.jks

Create the JAAS configuration

Create a file named jaas.conf with the following content:

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

Replace the placeholders with your instance credentials:

Create the Logstash pipeline configuration

Create a file named input.conf with the following content:

input {
    kafka {
        bootstrap_servers => "<your-internet-endpoint>"
        topics => ["<your-topic>"]

        security_protocol => "SASL_SSL"
        sasl_mechanism => "PLAIN"

        jaas_path => "<path-to-jaas-conf>"

        ssl_truststore_password => "KafkaOnsClient"
        ssl_truststore_location => "<path-to-truststore-jks>"

        ssl_endpoint_identification_algorithm => ""

        group_id => "<your-group-id>"
        consumer_threads => 3
        auto_offset_reset => "earliest"
    }
}

output {
    stdout {
        codec => rubydebug
    }
}

Replace the placeholders with your values:

The following parameters have fixed values specific to ApsaraMQ for Kafka internet connections:

Optional parameters: