Docs

Message Queue for Kafka

Best Practices

Kafka producer best practices

This document uses the official Confluent Java SDK as an example to provide recommendations for using Kafka producers and consumers. Before using Message Queue for Kafka to produce and consume messages, it is recommended that you read the following recommendations to improve access efficiency and business stability.

Message ordering

Messages in a VolcanoEngine Kafka instance are guaranteed to be in first-in, first-out order within the same partition. When messages are written to the same partition, if message A is written before message B, message A is guaranteed to be read by the client before message B. Note that here, message ordering is guaranteed only for messages sent sequentially by the same producer. For messages sent by different producers, ordering cannot be guaranteed because the arrival sequence at the server cannot be determined.

Based on the above characteristics, if you want to implement message ordering, you can consider the following approaches:

Global ordering: Create a Topic with only one partition. Because the Topic has only one partition, the order of messages received is strictly consistent with the order in which the producer client sends them. However, from the perspective of a single-partition topic, neither message writing nor reading can fully utilize the complete performance of the cluster. Only when multiple single-partition topics are used simultaneously is it possible to maximize the cluster's performance.

**Partition ordering:** Messages within a Kafka partition are inherently ordered, so message ordering can be achieved by writing messages for which order must be preserved to the same partition. Applicable to scenarios where message ordering is not required for all messages, or where ordering is required only for specific categories of messages.

A Topic with a single partition has significant limitations in terms of production and consumption performance. In practice, it is recommended to implement business logic using partition-based ordering. To achieve ordering for similar messages, write messages that require ordering into the same partition.

Message reliability

The acks configuration defines the method for acknowledging message writes and supports the following three configurations:

acks=0: The result of message writing is not tracked; the server does not return any result for this message, regardless of whether the write succeeds or fails.

acks=1: After writing to the primary replica, the server can then return the write result to the producer client.

acks=-1 or acks=all: The write result is returned to the producer client only after the message has been written to both the primary and replica.

For the three types of acks configuration, from top to bottom, performance decreases sequentially while data reliability increases sequentially. We recommend that you directly use the most reliable configuration method.

In distributed systems, intermittent send failures may occur due to issues such as network problems or primary node switching. You can configure the number of retry attempts for write failures using the retries parameter; the default number of retries is the maximum value of a long integer. The retry interval can be configured using the retry.backoff.ms parameter, with a default interval of 100 ms. The recommended configuration is 3 retry attempts and a retry interval of 1000 ms.

Partition selection

In practice, when a message is written, a specific partition within the Topic is selected for writing. The partition selection logic is as follows:

When a message specifies a message key, the partition is selected based on the hash result of the message key. Messages with the same key are assigned to the same partition number. Therefore, when the same message key is assigned to a batch of messages, these messages can be ordered within partitions. Caution: After increasing the number of partitions for a Topic, the partition number selected by the message key may change.

For client versions earlier than 2.4, if the message key is not specified, messages are written to each partition in turn in a round-robin fashion. However, this approach may result in suboptimal client message aggregation, which can affect sending performance.

For client versions 2.4 and above, if no message key is specified, messages are written to partitions using sticky partition selection. This approach is primarily intended to address issues with unsatisfactory aggregation. When selecting a partition, messages are preferentially written to the same partition as the previous message until the client’s message aggregation for that partition reaches the configured batch size limit (batch.size). While ensuring the effectiveness of message aggregation, balanced writes across partitions are also achieved in the long run.

In addition to the default implementation above, users can also implement custom partition selection logic.

It is recommended to simply use the default partition selection logic. When there is no message key, the default logic already ensures balanced distribution of messages across partitions. In scenarios where message keys are used, reducing hash collisions for message keys can effectively distribute messages and prevent certain partitions from handling too many business operations and messages. When implementing custom partition selection, it is also important to ensure that partitions are selected as evenly as possible, in order to prevent business operations and messages from becoming overly concentrated in certain partitions.

Send result

The current implementation of producer message sending uses asynchronous logic. That is, when you call the send method to write a message, the actual message is only written to the local cache and is not sent to the server. Messages are aggregated by partition in the local cache, after which an asynchronous sending thread scans and sends the aggregated messages to the server.

The send method only writes to the buffer and does not indicate that the message has actually been written successfully. To obtain the actual result of a message write operation, you can choose from the following methods:

When you call the send method, bind a callback function Callback to each message.

In the producer, you can inject a custom interceptor that implements the ProducerInterceptor interface by configuring interceptor.classes. This interceptor passes the result or exception of message writing via the onAcknowledgement method.

The return value of the send method is a Future object. You can directly call the get method of this object to block and wait for the result of the message write operation.

Of the three methods described above, it is recommended to use either of the first two. The third method is actually a pseudo-synchronous implementation, which can severely affect production performance on the client side and is not recommended.

Production performance

Producers use memory caching and message aggregation to reduce network requests between producers and the server, thereby achieving a significant improvement in throughput performance. The following configuration options for the aggregation capability on the production side can be customized:

The batch.size configuration defines the maximum message size for a single aggregation. The default size is 16KB. Setting a smaller value is not recommended. If you find that an excessively high server-side request production rate is causing elevated server CPU usage, you can increase this value appropriately to enhance aggregation capability and reduce CPU consumption.

The linger.ms configuration defines the minimum aggregation time for a single aggregation. The default is 0 ms, which means messages are sent as quickly as possible. Appropriately adjusting this parameter can also increase the level of message aggregation. However, this will increase the latency of message delivery. In scenarios where message aggregation is poor and production requests increase significantly, you can also appropriately increase this parameter to choose a suitable balance between the degree of message aggregation and acceptable sending delay.

The send.buffer.bytes configuration defines the size of the TCP buffer on the sending side. The default value is 128 KB. If a client sends large individual messages or if batch.size is increased, you should adjust the TCP buffer size accordingly. A recommended range is between 128 KB and 1024 KB.

Message timestamp

Message timestamps support two different methods for populating, which are used in server-side cluster configuration. For a detailed explanation, see Modify parameter configuration.

LogAppendTime: use the time when the server writes the message as the message timestamp.

CreateTime: (default) Uses the time when the producer creates the message, which is the timestamp automatically attached when the message is written.

Message timestamps are used in scenarios such as calculating message expiration, aging, and other scenarios. Messages sent by the client must have reasonable timestamps. If a message timestamp is entered incorrectly, data may not be aged out or deleted at the expected time. After writing a message, you can troubleshoot by querying the message offset. Usually, older versions of the API may have issues with missing message timestamps. It is recommended to use the recommended client version.

By default, if the message timestamp is not specified, the Confluent SDK uses the producer's local current time. If you need to manually specify the time, you should pay attention to entering the correct timestamp to avoid affecting server-side business processes such as message aging.

Multithreading use

The producer is implemented in a thread-safe manner. Therefore, in client-side business implementations, it is recommended to use a producer pool to provide producers for different multithreaded business operations, thereby avoiding the creation of a separate producer for each business operation.

Graceful exit

The producer uses asynchronous sending. Before you plan to shut down the producer, it is recommended to call the producer's flush method to proactively push cached messages to the server, and then call the close method to close the producer client. Otherwise, some messages in the cache may not be released in time before shutdown, resulting in message loss.

Domain name resolution

VolcanoEngine Kafka instances are deployed as a distributed cluster, with the initial access point provided as a domain name. When the client connects using a domain name, it is recommended to configure the client's DNS resolution method to resolve all IP addresses, that is, client.dns.lookup=use_all_dns_ips. Ensure that clients can continue to operate normally when backend nodes in the distributed cluster change.

Last updated: 2025.09.16 10:31:24