Cloud Native Message Engine
Cloud Native Message Engine
Document
- Documentation
Cloud Native Message EngineBMQ Instance User GuideBest practicesBest practices for consumer
Copy page
Download PDFBest practices
Best practices for consumer
Copy pageDownload PDFBest practices for consumer
This document describes consumer best practices when using VolcanoEngine BMQ instances.
Basic concepts
Interaction process
In BMQ consumption scenarios, the interaction process between consumers and the server is shown in the following diagram.
What is a consumer
A client process that subscribes to and consumes data from a specific topic is called a consumer (Consumer). A consumer does not exist independently, but belongs to a consumer group (Consumer Group). A consumer group can contain multiple consumers. They jointly and concurrently consume data from the same topic, with each consumer responsible for processing data from only a subset of partitions. This design allows BMQ to scale horizontally by increasing the number of consumers when handling large-scale data streams, thereby improving consumption capacity.
What is a consumer group
A consumer group can include multiple consumers. Consumers belonging to the same consumer group subscribe to or consume the same topic, and each consumer, according to the allocation strategy, consumes only a subset of partitions within a topic. For example, if Topic A has 4 partitions and Consumer Group A contains only one consumer, Consumer A1, then Consumer A1 will consume data from all partitions of Topic A. If Consumer Group A has 2 consumers, Consumer A1 and Consumer A2, then Consumer A1 will consume data from 2 partitions of Topic A, and Consumer A2 will consume data from the other 2 partitions.
What is Kafka consumption progress
Each consumer group tracks its consumption progress by submitting offset information. The client supports two offset submission mechanisms: manual submission and periodic submission. For business scenarios that are sensitive to repeated consumption or message loss on the consumer side, the configuration can be adjusted according to business characteristics.
Consumption progress is usually indicated by Lag, calculated as Lag = MaxOffset (HW) - CommittedOffset, where:
- MaxOffset (HW): The latest offset recorded in the partition.
- CommittedOffset: The last offset submitted by the consumer group. Data before CommittedOffset has been successfully consumed by the consumer group.
When the server does not have offset information submitted by the consumer group, the BMQ client can specify the starting position for consumption by setting the auto.offset.reset policy, supporting consumption from either the earliest record or the latest record.
What is high-level consumption
The consumption coordination in the high-level consumption model is managed by the server-side Coordinator component. In the high-level consumption model, consumers do not need to decide which partitions to consume. The server assigns partitions based on the PARTITION_ASSIGNMENT_STRATEGY_CONFIG parameter configured by the client. The default value of this parameter is RangeAssignor, which evenly distributes partitions among all consumers in the group.
What is low-level consumption
The consumption coordination in the low-level consumption model is actively managed by the consumer. The low-level consumption model is typically suitable for business services that need to maintain state. Services in a specific state need to consume fixed partitions. In this scenario, consumers must actively specify the partitions they consume, without intervention from the server-side Coordinator component.
Recommended configuration for high-level consumption
Configuration recommendations
- Evaluate the processing time of each message: Avoid performing synchronous remote calls with unpredictable delays or complex, time-consuming resource processing when handling messages.
- Each consumer group should consume only one topic: BMQ allows a consumer group to consume multiple topics. In this scenario, any processing delay for topic data may cause the entire consumer group to rebalance. Therefore, it is recommended that each consumer group consumes only one topic to improve overall consumption stability and simplify issue diagnosis.
Parameter configuration recommendations:
Configuration example
//View the corresponding endpoint information in the console
String server = "xxx.";
//Topic to which the message applied for in the console belongs
String topic = "this is your topic.";
//consumerGroup for consuming messages applied for in the console
String group = "this is your group.";
//Offset consumption strategy: earliest, latest, none
String offsetReset = "earliest";
Properties properties = new Properties();
//Required parameters
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, group);
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offsetReset);
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//Tune parameters according to the scenario
//If set to false, manual offset commit is required, that is, call the consumer.commitSync() method.
//The default value is true, which means offsets are committed automatically. The current offset is committed to Kafka every 5 seconds.
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
//Heartbeat interval, that is, the interval in milliseconds at which the client sends a heartbeat to the server. The default value is 3 seconds.
properties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
//If the server does not receive a heartbeat within session.timeout.ms, the current session is considered expired and a rebalance will be triggered.
//It is recommended to increase this value appropriately, for example, set it to 60 seconds.
properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 60000);
//The maximum number of records returned in a single poll call. The default value is 500; it is recommended to reduce this value, for example, to 100.
properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
//Used to specify the maximum interval between two poll calls. The default is 5 minutes. If this interval is exceeded, a rebalance will be triggered.
properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
List<String> topicList = Lists.newArrayList(topic);
consumer.subscribe(topicList);
while (true) {
try {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, String> record : records) {
logger.info("consumed record, topic={}, partition={}, offset={}, key={}, value={}",
record.topic(),
record.partition(),
record.offset(),
record.key(),
record.value());
}
} catch (Exception e) {
logger.error("consume error", e);
}
}
Recommended configuration for low-level consumption
Configuration recommendations
- Start a single consumer in each process: It is recommended to run a single consumer instance in each process so that, when analyzing consumer behavior or troubleshooting issues, you can use trace and dump operations to obtain the status of relevant objects or call stack information.
- Avoid long periods without calling the poll method: Ensure that the consumer calls the poll method regularly. It is recommended to call the poll method every few seconds to prevent the consumer from failing to read data returned by the server from the socket buffer in a timely manner due to prolonged periods without calling the poll method, which may cause blocking of server write operations.
- max.poll.records parameter: The maximum number of messages retrieved in a single poll call. Reducing the number of messages can shorten the interval between poll calls and promote frequent interactions at the network level. The default value for this parameter is 500. You may decrease it appropriately based on your actual business scenario or consumer capacity.
Configuration example
// View the corresponding endpoint information in the console
String server = "xxx.";
// Topic to which the message applied for in the console belongs
String topic = "this is your topic.";
// consumerGroup for consuming messages applied for in the console
String group = "this is your group.";
// Partition value partition
int partition = 0;
Properties properties = new Properties();
// Required parameter
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, group);
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offsetReset);
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
// Tune parameters according to the scenario
// If set to false, manual offset submission is required, that is, call the consumer.commitSync() method.
// The default value is true, which means offsets are automatically committed. The current consumed offset will be committed to Kafka every 5 seconds.
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
// The maximum number of records returned by a single poll method call. The default value is 500. It is recommended to reduce this value, for example, to 100.
properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
// Partition assignment. Different nodes, processes, or threads need to assign different partitions.
consumer.assign(Collections.singletonList(new TopicPartition(topic, partition)));
while (true) {
try {
// Poll method latency. ...
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
logger.info("consumed record, topic={}, partition={}, offset={}, key={}, value={}",
record.topic(),
record.partition(),
record.offset(),
record.key(),
record.value());
}
} catch (Exception e) {
logger.error("consume error", e);
}
}
Common parameter descriptions
tip
The default values for the following parameters are based on Java SDK version 2.3.0 and may differ in SDKs for other programming languages or versions.
FAQ
Frequent consumer rebalances
A common reason for consumer rebalance is that the server does not receive a heartbeat from the consumer within the session.time.out period, determines that the consumer has left the consumer group, and triggers a rebalance.
You can reduce or avoid this by adjusting the following parameters:
- session.timeout.ms: Increase this parameter appropriately. The adjustment logic is greater than (max.poll.records × processing time per message), ensuring that the consumer can call the poll method at least once within the heartbeat timeout period to trigger the heartbeat thread and send a heartbeat.
- max.poll.records: Decrease this parameter appropriately. The adjustment logic is less than (session.timeout.ms × number of messages the client can process per second).
How to improve consumer capacity
If the consumer capacity cannot keep up with the message production rate, Lag will continue to increase.
You can adjust consumer capacity in the following ways:
- Scale out the number of consumer instances
- The number of consumer instances can be scaled out to match the number of partitions. If you have already scaled out to match the number of partitions, you can further increase the number of consumer instances by increasing the number of partitions.
- Process messages asynchronously after retrieval
- One of the main reasons for slow consumption speed is insufficient data processing capability. Adopting an asynchronous processing mechanism can prevent the data fetching process from being blocked. Please note that asynchronous processing itself does not optimize the existing data processing logic. To improve processing efficiency, you can increase the concurrency of asynchronous processing.
- Concurrent processing after retrieving messages
- In most cases, a single poll call retrieves multiple messages. The overall consumption capability can be improved through concurrent processing mechanisms.
- Improve the processing capability of dependent services
- If data processing depends on other downstream services, the consumption capability can be enhanced by improving the processing capability of downstream dependencies.
Excessive data volume fetched over the public network leads to connection termination
The default value of fetch.max.bytes is 50MB. When consumers consume over the public network, if the purchased public network bandwidth is less than 50MB, the connection between the consumer and the server may be terminated due to bandwidth limitations.
Adjust the following parameters to limit the amount of data the consumer retrieves each time:
- fetch.max.bytes: Adjust to less than the public network bandwidth. In some test scenarios, this parameter can be set to a smaller value, such as 10K (10240 bytes).
- max.partition.fetch.bytes: It is recommended to set it to one fourth of the value of the fetch.max.bytes parameter.
Last updated: 2026.01.26 15:45:56
All rights reserved © Beijing Volcano Engine Technology Co., Ltd2024
