ApsaraMQ for Kafka:Connect a ApsaraMQ for Kafka instance to Filebeat as an output

Log on to the ApsaraMQ for Kafka console.

In the Resource Distribution section of the Overview page, select the region where your instance is deployed.

On the Instances page, click the name of the instance that you want to connect to Filebeat as an output.

On the Instance Details page, obtain an endpoint of the instance in the Endpoint Information section. In the Configuration Information section, obtain the values of the Username parameter and Password parameter.

Note

For information about the differences among endpoints, see Comparison among endpoints.

Log on to the ApsaraMQ for Kafka console.

In the Resource Distribution section of the Overview page, select the region where your instance is deployed.

On the Instances page, click the name of the instance that you want to manage.

In the left-side navigation pane, click Topics.

On the Topics page, click Create Topic.

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

Parameter	Description	Example
Name	The topic name.	demo
Description	The topic description.	demo test
Partitions	The number of partitions in the topic.	12
Storage Engine Note You can select the type of the storage engine only if you use a Professional Edition instance. If you use a Standard Edition instance, cloud storage is selected by default.	The type of the storage engine that is used to store messages in the topic. ApsaraMQ for Kafka supports the following types of storage engines: Cloud Storage: If you select this value, the system uses Alibaba Cloud disks for the topic and stores data in three replicas in distributed mode. This storage engine features low latency, high performance, durability, and high reliability. If you set the Instance Edition parameter to Standard (High Write) when you created the instance, you can set this parameter only to Cloud Storage. Local Storage: If you select this value, the system uses the in-sync replicas (ISR) algorithm of open source Apache Kafka and stores data in three replicas in distributed mode.	Cloud Storage
Message Type	The message type of the topic. Valid values: Normal Message: By default, messages that have the same key are stored in the same partition in the order in which the messages are sent. When a broker in the cluster fails, the order of the messages may not be preserved in the partitions. If you set the Storage Engine parameter to Cloud Storage, this parameter is automatically set to Normal Message. Partitionally Ordered Message: By default, messages that have the same key are stored in the same partition in the order in which the messages are sent. When a broker in the cluster fails, the messages are still stored in the partitions in the order in which the messages are sent. Messages in some partitions cannot be sent until the partitions are restored. If you set the Storage Engine parameter to Local Storage, this parameter is automatically set to Partitionally Ordered Message.	Normal Message
Log Cleanup Policy	The log cleanup policy that is used by the topic. If you set the Storage Engine parameter to Local Storage, you must configure the Log Cleanup Policy parameter. You can set the Storage Engine parameter to Local Storage only if you use an ApsaraMQ for Kafka Professional Edition instance. ApsaraMQ for Kafka provides the following log cleanup policies: Delete: The default log cleanup policy. If sufficient storage space is available in the system, messages are retained based on the maximum retention period. After the storage usage exceeds 85%, the system deletes messages from the earliest stored message to ensure service availability. Compact: The Apache Kafka log compaction policy is used. For more information, see Kafka 3.4 Documentation Log compaction ensures that the latest values are retained for messages that have the same key. This policy is suitable for scenarios such as restoring a failed system or reloading the cache after a system restarts. For example, when you use Kafka Connect or Confluent Schema Registry, you must store the information about the system status and configurations in a log-compacted topic. Important You can use log-compacted topics only in specific cloud-native components such as Kafka Connect and Confluent Schema Registry. For more information, see aliware-kafka-demos.	Compact
Tag	The tags that you want to attach to the topic.	demo

Run the cd command to switch to the installation directory of Filebeat.

Create a configuration file named output.conf.

1. Run the vim output.conf command to create an empty configuration file.

2. Press the I key to enter the insert mode.

3. Enter the following content:

   filebeat.inputs:
   - type: stdin
   
   output.kafka:
     hosts: ["alikafka-pre-cn-zv**********-1-vpc.alikafka.aliyuncs.com:9092", "alikafka-pre-cn-zv**********-2-vpc.alikafka.aliyuncs.com:9092", "alikafka-pre-cn-zv**********-3-vpc.alikafka.aliyuncs.com:9092"]
   
     topic: 'filebeat_test'
   
     required_acks: 1
     compression: none
     max_message_bytes: 1000000

   Parameter	Description	Example
   hosts	The VPC endpoint of the ApsaraMQ for Kafka instance. ApsaraMQ for Kafka supports the following VPC endpoints: Default endpoint Simple Authentication and Security Layer (SASL) endpoint	alikafka-pre-cn-zv**********-1-vpc.alikafka.aliyuncs.com:9092, alikafka-pre-cn-zv**********-2-vpc.alikafka.aliyuncs.com:9092, alikafka-pre-cn-zv**********-3-vpc.alikafka.aliyuncs.com:9092
   topic	The name of the topic.	filebeat_test
   required_acks	The reliability level of acknowledgments (ACK). Valid values: 0: no response 1: waits for local commit -1: waits for all replicas to commit Default value: 1.	1
   compression	The data compression codec. Default value: gzip. Valid values: none: none snappy: the C ++ development package used for compression and decompression lz4: the lossless data compression algorithm that enables faster compression and decompression gzip: the file compression program for GNU free software	none
   max_message_bytes	The maximum size of a message. Unit: bytes. Default value: 1000000. The value must be smaller than the maximum message size that you specify for ApsaraMQ for Kafka.	1000000

   For more information about parameter settings, see Kafka output plugin.

4. Press the Esc key to return to the command line mode.

5. Press the : key to go to the bottom line. Enter wq, press the Enter key to save the file, and then exit.

Send a message to the topic that you created.

1. Run the ./filebeat -c ./output.yml command.

2. Enter test and press Enter.

Log on to the ApsaraMQ for Kafka console.

In the Resource Distribution section of the Overview page, select the region where your instance is deployed.

On the Instances page, click the name of the instance that you want to manage.

In the left-side navigation pane, click Topics.

On the Topics page, find the topic whose partition status you want to view, and choose More > Partition Status in the Actions column.

Table 1. Information about the status of a partition

Parameter	Description
Partition ID	The ID of the partition.
Minimum Offset	The earliest offset based on which messages in the partition are consumed.
Maximum Offset	The latest offset based on which messages in the partition are consumed.
Last Updated At	The most recent point in time when a message is stored in the partition.

Log on to the ApsaraMQ for Kafka console.

In the Resource Distribution section of the Overview page, select the region where your instance is deployed.

On the Instances page, click the name of the instance that you want to manage.

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

On the Message Query page, select Search by offset from the Search Method drop-down list.

Select a topic name from the Topic drop-down list and a partition from the Partition drop-down list, enter an offset value in the Offset field, and then click Search.