root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-18881 Document the ConsumerRecord as non-thread safe (#19056) 

There are 3 issues (at least) about the multithreaded issue on ConsumerRecords. Hence, it would be better to document it completely. 

Reviewers: Kirk True <ktrue@confluent.io>, TengYao Chi <kitingiao@gmail.com>, Ken Huang <s7133700@gmail.com>, Xuan-Zhang Gong <gongxuanzhangmelt@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
This commit is contained in:
Logan Zhu 2025-03-03 13:03:36 +08:00 committed by GitHub
parent 2e6e5304c0
commit bf660fdeb6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 21 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecord.java
View File

@ -21,12 +21,31 @@ import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.record.TimestampType;
import java.util.ConcurrentModificationException;
import java.util.Optional;
/**
* A key/value pair to be received from Kafka. This also consists of a topic name and
* a partition number from which the record is being received, an offset that points
* to the record in a Kafka partition, and a timestamp as marked by the corresponding ProducerRecord.
* <p>
*
* <h3>Thread Safety</h3>
* This consumer record is <b>not thread-safe</b>. Concurrent access to a {@code ConsumerRecord} instance by
* multiple threads may result in undefined behavior, including but not limited to the following:
* <ul>
* <li>Throwing {@link ConcurrentModificationException} (e.g., when concurrently modifying {@link #headers()}).</li>
* <li>Data corruption or logical errors (e.g., inconsistent state of {@code headers} or {@code value}).</li>
* <li>Visibility issues (e.g., modifications by one thread not being visible to another thread).</li>
* </ul>
*
* <p>
* In particular, the {@link #headers()} method returns a mutable collection of headers. If multiple
* threads access or modify these headers concurrently, it may lead to race conditions or inconsistent
* states. It is the responsibility of the user to ensure that multi-threaded access is properly synchronized.
*
* <p>
* Refer to the {@link KafkaConsumer} documentation for more details on multi-threaded consumption and processing strategies.
*/
public class ConsumerRecord<K, V> {
public static final long NO_TIMESTAMP = RecordBatch.NO_TIMESTAMP;