Docs

Torch Log Service

Data consumption and shipping

Consume logs via Kafka protocol

TLS provides log consumption via the Kafka protocol. That is to say, one log topic can be used as a Kafka topic for consumption. This topic describes the steps to consume log data using the Kafka protocol.

Background information

TLS supports enabling log consumption via the Kafka protocol for specified log topics. Once enabled, you can consume the log topic as a Kafka topic, with each log corresponding to a Kafka message. In actual business scenarios, after integrating TLS with an open-source Kafka SDK, you can use the Kafka Consumer to ship logs collected from designated log topics to downstream big data components or data warehouses. This is suitable for streaming analytics and big data storage.

When consuming logs via the Kafka protocol, you can consume data as a consumer or a consumer group. Cross-project consumption is not supported.

Limits

Kafka protocol consumption supports Kafka Client versions 0.11.x to 2.0.x.

When the Kafka protocol consumption feature is enabled, you can consume log data collected on the server side during the runtime of the Kafka Consumer.

Log data can be consumed using the standard open-source Kafka Java SDK. For sample code, see Code examples. You can also use the Kafka plugins for Spark Streaming or Flink to integrate with TLS. For detailed instructions, see Consume logs using Spark Streaming and Consume logs using Flink.

To ensure secure log transmission, you must use the SASL_SSL connection protocol. The corresponding username is your TLS project ID, and the password is Volcanoengine account access key. For more details, see Code examples.

If a log topic contains multiple shards, orderly consumption can not be guaranteed. We recommend uploading logs in load balancing mode.

Billing description

Log consumption generates private or public network read traffic. For pricing information, see Billing items based on pay-by-feature mode.

Private network read traffic: When consuming log data via Kafka protocol to other Volcengine private network services, if both the log topic and the consumer are in the same region, you can use private transmission. This incurs a private network read traffic fee. For example, if source data is located in a log topic of the TLS Beijing region, and log data is consumed using Kafka to an ECS self-managed program, the corresponding private network traffic fee will be charged.

Public network read traffic: When consuming log data via Kafka protocol to external programs on the public network, you must use public transmission. This generates public network read traffic. For example, if source data is in a TLS log topic and log data is consumed via Kafka to a self-developed program in a self-built IDC, public network read traffic will be generated.

Tip

If the source log topic and the consumer end are in different regions, you must use public network transmission, which incurs public network read traffic.

Prerequisites

Log service has been enabled; a project and a topic are created; log data has been successfully collected. For more details, see Overview.

Ensure that your current account has permission to enable log consumption via Kafka protocol, i.e., the permission of Action ConsumeLogs. For more details, see Authorized actions.

Steps

1 Enable Kafka consumption

Before using various tools to consume log data via Kafka protocol, you need to enable Kafka consumption for the specified log topic.

Select the region where log service is located from the top navigation bar.

On the left navigation bar, select Resource Management > Project .

On the Project page, locate the specified project and click the project name.

In the Log topic section, find the specified log topic in the log topic list.

Click the Kafka protocol consumption switch in the log topic list.

Click the log topic name and select ... > Configure log consumption.

In the pop-up dialog box, enable Kafka consumption protocol configuration, and then click OK.

Parameters	Description
Compression type	Configure the compression type for consumed data; supported options include lz4, snappy, gzip, and no compression.
Consumptionformat	Configure the format of consumed data. Raw: Consume data in raw format. JSON: Consume data in JSON format; field-specific consumption is also supported. Escape mode: No escaping: output data as collected by log collection. Escaping: add escape characters to values of top-level nodes. Original log: { "a": "aa", "b": { "b1": "b1b1", "c1": "c1c1" } } Consumption result： No escaping: { "a": "aa", "b": { "b1": "b1b1", "c1": "c1c1" } } Escaping: { "a": "aa", "b": "{\"b1\":\"b1b1\", \"c1\":\"c1c1\"}" } Consumable log fields: You can customize fields to consume. By default, all fields are selected. Supported fields are as follows: __content__(can not be canceled), __source__, __path__, __context_flow__, tags.

After successfully enabling the Kafka protocol consumption feature, the details page of this log topic will display consumption parameters like username, hosts, topics. etc.

You can enable the Kafka protocol consumption feature again after it has been disabled for 2 minutes.

2 Consume logs via Kafka protocol

Currently, TLS supports log consumption through the Kafka Java SDK or Kafka Connector plugins used with frameworks like Spark and Flink. You can refer to the following section to configure basic Kafka parameters and review sample code for log data consumption.

Tip

Kafka protocol consumption feature takes about 30 seconds to start. We recommend that you begin consuming logs 30 seconds after enabling the feature.

Parameters description

When consuming logs via the Kafka protocol, you need to configure the following parameters, parameter details or format can be found in the details page of log topic.

Parameters	Description
Connection type	To ensure secure log transmission, you must use the SASL_SSL connection protocol. The corresponding username is the project ID of TLS. The password is the Volcengine account secret. For details, see Code examples.
username	Kafka SASL username. Set this to TLS project ID.
password	Kafka SASL user password. Use Volcanoengine account key. The format is ${access-key-id}#${access-key-secret}, where: ${access-key-id} should be replaced with your AccessKey ID. ${access-key-secret} should be replaced with your AccessKey Secret. Tip We recommend using the AK of the IAM user , and it must have Action ConsumeLogs permission. For more details, see Authorized actions.
hosts	The initial connection cluster address is formatted as service address:port number, for example tls-cn-beijing.ivolces.com:9093. Where: The service address is the entry point for TLS in the current region. Select the correct service entry point based on region and network type. For details, see Service address. The port number is fixed at 9093. Tip There is no need to specify https:// in the service address part of hosts.
topic	The Kafka protocol topic ID is formatted as out-log topic ID. For example, out-0fdaa6b6-3c9f-424c-8664-fc0d222c****. You can also view and copy the Kafka protocol topic ID from the Topic details page in the TLS console.

Error message

If uploading logs using Kafka protocol fails, error messages will be returned based on Kafka error code. For more information, see Kafka error list.

Additionally, TLS encapsulates authentication and configuration-related error information in the Java Kafka error code SASLAuthenticationException, as described below:

Error message	Description
invalid SASL/PLAIN request: expected 3 tokens.	User or password is not configured. For correct configuration, see Parameters description.
invalid SASL/PLAIN request: empty projectId.	The user field is not configured. For correct configuration, see Parameters description.
invalid SASL/PLAIN request: Password format wrong	Password field configuration error. For correct configuration, see Parameters description.
invalid SASL/PLAIN request: empty AccessKey.	AK is not configured. For correct configuration, see Parameters description.
invalid SASL/PLAIN request: empty SecretKey.	SK is not configured. For correct configuration, see Parameters description.
invalid SASL/PLAIN request: Invalid projectId.	The specified Project ID does not exist. Check if the Project ID is entered correctly.
Not allow consume, please open kafka consumer.	The specified topic does not have the consumption feature enabled. For instructions, see 1 Enable Kafka consumption.
Access Denied. You are not authorized to perform this operation.	The current user does not have permission for this action. We recommend checking user permissions.

Code examples

Consume logs with Kafka Java SDK

With simple parameter configuration, you can use various Kafka Consumer SDKs to consume data collected in TLS. The dependencies and sample code for consuming logs using the Kafka Java SDK are as follows:

Add dependencies.

Add the relevant kafka-clients dependency in the pom file.

<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.2.2</version>
</dependency>

Consume logs.

Refer to the following sample code to consume logs using Kafka Java SDK.

package org.kafka;

import java.util.*;
import java.util.concurrent.CountDownLatch;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaConsumeTest implements Runnable {

private final KafkaConsumer<String, String> consumer;
private ConsumerRecords<String, String> msgList;
private final String topic;
private static final String GROUPID = "MY_GROUP_ID";
private KafkaAdminClient adminClient;
private String consumeName;

public KafkaConsumeTest(String topicName,String consumeName) {
// consumeName
this.consumeName = consumeName;
// projectid
String userName = "${projectId}";
// Volcanoengine account secret key, or a sub-account key with the corresponding permissions. STS temporary security tokens are not supported.
String passWord = "${access-key-id}#${access-key-secret}";
Properties props = new Properties();
props.put("bootstrap.servers", "${hosts}"); // Consumption address. See documentation for details.
props.put("group.id", GROUPID);// Consumer group name; length: 1–256 characters.
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("auto.offset.reset", "earliest");
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
props.put(SaslConfigs.SASL_JAAS_CONFIG,
"org.apache.kafka.common.security.plain.PlainLoginModule " +
"required username=\"" + userName + "\" password=\"" + passWord + "\";");

this.consumer = new KafkaConsumer<String, String>(props);
this.topic = topicName;
this.consumer.subscribe(Arrays.asList(topic));
this.adminClient = (KafkaAdminClient) KafkaAdminClient.create(props);
}

@Override
public void run() {
int messageNo = 1;
System.out.println("---------Start consuming---------");
try {
while(true) {
msgList = consumer.poll(10);

if(null!=msgList&&msgList.count()>0){
for (ConsumerRecord<String, String> record : msgList) {
System.out.println(this.consumeName+"==="+messageNo+" offset==="+record.offset()+"=======receive: key = " + record.key() + ", value = " +
record.value());
messageNo++;
}
} else{
Thread.sleep(10);
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
consumer.close();
}
}

public static void main(String args[]) throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(1);
// ${out-TopicID} refers to the Kafka protocol consumption topic ID. The format is out+log topic ID, for example: "out-0fdaa6b6-3c9f-424c-8664-fc0d222c****".
KafkaConsumeTest test1 = new KafkaConsumeTest("${out-TopicID}","Consume1");
Thread thread1 = new Thread(test1);
thread1.start();
countDownLatch.await();
}
}

Consume logs using the Kafka Go SDK.

Run the following command to install Sarama.

go get github.com/Shopify/sarama@v1.38.1

Import the required dependencies such as Sarama, and consume logs.

Refer to the following sample code to use the Kafka Go SDK for log consumption.

package kafka_consume_example

import (
"context"
"fmt"

"github.com/Shopify/sarama"
)

type exampleConsumerGroupHandler struct{}

func (exampleConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (exampleConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h exampleConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
fmt.Printf("Message topic:%q partition:%d offset:%d message:%s\n", msg.Topic, msg.Partition, msg.Offset, msg.Value)
sess.MarkMessage(msg, "")
sess.Commit()
}
return nil
}

func ExampleConsumerGroup() {
config := sarama.NewConfig()
config.Version = sarama.V2_0_0_0 // specify appropriate version
config.ApiVersionsRequest = true
config.Consumer.Return.Errors = true
config.Net.SASL.Mechanism = "PLAIN"
config.Net.SASL.Version = int16(0)
config.Net.SASL.Enable = true
config.Net.TLS.Enable = true
config.Consumer.Offsets.Initial = sarama.OffsetNewest
// projectId
config.Net.SASL.User = "${projectId}"
// Volcanoengine account secret key, or a sub-account key with the corresponding permissions. STS temporary security tokens are not supported.
config.Net.SASL.Password = "${access-key-id}#${access-key-secret}"
// host is the consumption address. See documentation for details. Replace my-group with the actual consumer group name; length: 1–256 characters.
group, err := sarama.NewConsumerGroup([]string{"${hosts}"}, "my-group", config)
if err != nil {
panic(err)
}
defer func() { _ = group.Close() }()

// Track errors
go func() {
for err := range group.Errors() {
fmt.Println("ERROR", err)
}
}()

// Iterate over consumer sessions.
ctx := context.Background()
for {
// ${out-TopicID} is the Kafka protocol consumption topic ID, formatted as out+log topic ID, for example: "out-0fdaa6b6-3c9f-424c-8664-fc0d222c****".
topics := []string{"${out-TopicID}"}
handler := exampleConsumerGroupHandler{}

// `Consume` should be called inside an infinite loop, when a
// server-side rebalance happens, the consumer session will need to be
// recreated to get the new claims
err := group.Consume(ctx, topics, handler)
if err != nil {
panic(err)
}
}
}

Related operations

View consumer group

After consuming log data using the Kafka Java SDK or frameworks such as Spark, Flink, and the Kafka Connector plugin, you can view consumer group information in the TLS console.

Select the region where log service is located from the top navigation bar.

In the left navigation bar, select Data processing > Log consumption.

On the Kafka tab, locate and expand the project linked to the target Kafka consumer group, and then click the specified Kafka consumer group.

View consumption information for the target consumer group.

Reset consumption checkpoint.

Notice

Before resetting the consumption checkpoint, stop the corresponding consumer group’s process. Once the reset is complete, restart the relevant consumer process.

After consuming log data through Kafka Java SDK or Kafka Connector plugins for frameworks like Spark and Flink, you can reset the consumption checkpoint in the TLS console. You can also reset the consumption checkpoint to consume historical data.

Select the region where log service is located from the top navigation bar.

In the left navigation bar, select Data processing > Log consumption.

On the Kafka tab, locate and expand the project linked to the target Kafka consumer group, and then click the specified Kafka consumer group.

Reset consumption checkpoint.

Click Reset consumption checkpoint in the upper right corner of the Log topic List tab, or click Reset consumption checkpoint in the Actions column for the specified shard.

Select reset checkpoint.

Reset location	Description
Earliest location	Consume data starting from the earliest entry in the log topic.
Latest location	Skip all historical data and consume directly from the most recent entry written in the log topic.
Specified time point	Consume data starting from a specified time point in the past; the available time range depends on the Log storage period of the log topic.

Click Reset.

View monitoring dashboard

After consuming log data via the Kafka Java SDK or frameworks such as Spark, Flink, and their Kafka Connector plugins, you can monitor the latency and speed trends of consumer groups in the log service console.

Select the region where log service is located from the top navigation bar.

In the left navigation bar, select Data processing > Log consumption.

On the Kafka tab, locate and expand the project linked to the target Kafka consumer group, and then click the specified Kafka consumer group.

On the Monitoring dashboard tab, view consumption-related dashboards.

The monitoring dashboard displays consumption latency, consumption rate, and related trends for the Kafka consumer group. You can monitor Kafka protocol consumption backlog in real time through the dashboard. For specific operations, see Visualization of Kafka protocol consumption metrics.

Last updated: 2026.07.13 11:49:40