Self-managed Apache Kafka

Apache Kafka version and configuration

• Your Apache Kafka version must be 0.10.0 or later.

• The advertised.listeners parameter on the Apache Kafka server must be configured with an IP address. For example: advertised.listeners=PLAINTEXT://192.168.XX.XX:9092.

Procedure

1. Log on to the EventBridge console. In the left-side navigation pane, click Event Streams.

2. In the top navigation bar, select a region and click Create Event Stream.

3. In the Create Event Stream panel, set the Task Name and Description, configure the following parameters, and then click Save.

   • Create task

     1. In the Source step of the wizard, set Data Provider to Self-managed Service > Apache Kafka, configure the following parameters, and then click Next.

     Parameter	Description	Example
     Endpoint	A Kafka cluster broker endpoint consists of the broker's IP address and port number in the format {Broker IP address}:{port number}.	192.0.XX.XX:9093,198.51.XX.XX:9093,203.0.XX.XX:9093
     Topic	The name of the topic.	testTopic
     Group ID	The ID of the consumer group that subscribes to the topic. Note This is the name of the Kafka consumer group. To avoid disrupting existing services, use a new, dedicated group ID for the event source. Do not reuse a group ID from other applications.	GID_TEST
     Network	Select the network type. VPC Internet	Internet
     VPC	Select the VPC ID.	vpc-bq1huohcvuo****
     vSwitch	Select the vSwitch ID.	vsw-bqu1hdguoo****
     Security Group	Select the security group of the Kafka instance.	sg-dguigreuohpnv****
     Authentication method	PLAINTEXT SASL_PLAINTEXT Select the authentication method. PLAINTEXT SASL_PLAINTEXT Username: Enter the SASL username. Password: Enter the SASL password. SASL Mechanism: The SASL authentication mechanism. Valid values: PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512.	PLAINTEXT
     Consumer Offset	Select the offset from which to start consuming messages. Latest Offset Earliest Offset	Latest Offset
     Data Format	Specifies how EventBridge encodes binary data. If you do not have specific encoding requirements, select Json. Json (default): Encodes binary data as a UTF-8 JSON string in the payload. Text: Encodes binary data into a string by using UTF-8 and places it in the payload. Binary: Encodes binary data into a Base64 string and places it in the payload.	Json (default)
     Messages	The maximum number of messages per batch. A batch is sent when it reaches this size. The value must be in the range of 1 to 10,000.	100
     Interval (Unit: Seconds)	The maximum time to wait before sending a batch, in seconds. If a batch is not full, it is sent when this interval is reached. A value of 0 means messages are sent immediately.	3

     1. In the Filtering, Transformation, and Sink steps, configure the event filtering rules, transformation rules, and event sink. For more information about how to configure event transformation, see Use Function Compute to perform data cleansing.

   • Task properties

     Configure the retry policy and dead-letter queue for the event stream. For more information, see Retries and dead-letter queues.

4. Return to the Event Streams page, click the name of the target event stream, and then click Enable in the upper-right corner of the Overview page.

   After you enable the event stream, it starts running within 30 to 60 seconds. You can view the status in the Status column on the Event Streams page.

Event example

{
    "specversion": "1.0",
    "id": "8e215af8-ca18-4249-8645-f96c1026****",
    "source": "apachekafka",
    "type": "apachekafka:Topic:Message",
    "subject": "apachekafka:192.0.XX.XX:9093,198.51.XX.XX:9093,203.0.XX.XX:9093:topic:****",
    "datacontenttype": "application/json; charset=utf-8",
    "time": "2022-06-23T02:49:51.589Z",
    "aliyunaccountid": "182572506381****",
    "data": {
        "topic": "****",
        "partition": 7,
        "offset": 25,
        "timestamp": 1655952591589,
        "headers": {
            "headers": [
                {
                    "key": "head01",
                    "value": "this is my header"
                }
            ],
            "isReadOnly": false
        },
        "key": "keytest",
        "value": "hello kafka msg"
    }
}

For the definitions of the parameters specified by the CloudEvents specification, see Event overview.

The following table describes the parameters in the data field.