TLS provides Kafka protocol consumption functionality. You can use Flink's flink-connector-kafka plugin to connect to TLS and consume data from TLS into downstream big data components or data warehouses.

Scenario overview

Apache Flink is a distributed processing engine and framework for stateful computation on bounded and unbounded data streams. Flink provides the Apache Kafka connector (flink-connector-kafka) to read and write data in Kafka topics. TLS supports enabling Kafka protocol consumption functionality for specified log topics. Once enabled, Flink can consume log topics as Kafka topics, such as consuming them into downstream big data components or data warehouses. This is suitable for streaming computation or big data storage scenarios.

Prerequisites

A log project and log topic are created. For detailed instructions, see .

Kafka protocol consumption functionality has been enabled for the specified log topic, and the Kafka protocol consumption topic ID has been obtained. For more details, see .

We recommend using IAM users for access authorization. Before using an IAM user, ensure that the Volcanoengine primary account has created an IAM user and granted consumption-related permissions. For more details, see .

Access Key for the current login account has been obtained. For more information, see Create Access Key.

Notice

Apache Flink has multiple built-in Kafka clients, and the client versions used may vary between different Flink distributions. Currently, TLS only supports Kafka client connections of version 0.11.x and above. The corresponding Flink version is 1.7.x and above. If you use Flink below 1.7.x, you need to manually specify the Kafka client version as 0.11.x or above.

Configuration steps

1. Add Maven dependencies

When consuming log data from the TLS log service of Volcano Engine using Flink, you need to use Flink's flink-connector-kafka as the connector.

Add the relevant Maven dependencies for flink-connector-kafka to your project.

The dependency details are as follows:

2. Configure Kafka source for Flink

The Kafka Source provides builder classes to create instances of KafkaSource. For usage and implementation details, see Flink.

When building KafkaSource, you must specify the basic properties using the following methods.

Method	Description
setBootstrapServers()	The bootstrap server should be configured as the cluster address for the initial connection. The format is service address:port, for example tls-cn-beijing.ivolces.com:9093. The service address is the entry point for TLS in the current region. Select the correct service address based on the region and network type, for more information seeService address. The port number is fixed at 9093.
setTopics()	The Kafka topic to subscribe to. This should be configured as the Kafka protocol consumption topic ID of TLS, in the format out-log topic ID, for example out-0fdaa6b6-3c9f-424c-8664-fc0d222c****. You can view and copy the Kafka protocol consumption topic ID on the Topic details page in the TLS console.
setGroupId()	Consumer group ID.
setValueOnlyDeserializer()	The deserializer for parsing Kafka messages.
setProperty()	Settings for security mode, authentication mode, and more. The following configurations must be specified: CommonClientConfigs.SECURITY_PROTOCOL_CONFIG must be set to SASL_SSL. SaslConfigs.SASL_MECHANISM must be set to PLAIN. Configure the authentication information of the current logged-in user in SaslConfigs.SASL_JAAS_CONFIG, including: useranme： Kafka SASL username, in the format ${project-id}#${net}. ${project-id}: Replace with the project ID of the project to which the log topic belongs. ${net}: Replace with the specific network type. public: Public network type, such as . volces.com. private: Private network type, such as . ivolces.com. password： Kafka SASL user password, in the format ${access-key-id}#${access-key-secret}. ${access-key-id}: Replace with the AccessKey ID of your Volcano Engine account. ${access-key-secret}: Replace with the AccessKey Secret of your Volcano Engine account.

The following code snippet demonstrates how to construct KafkaSource to consume log data from the TLS log topic "0fdaa6b6-3c9f-424c-8664-fc0d222c****".

Torch Log Service