Kevin Wu
7ac40f4ded
When the metadata layer (e.g. MetadataLoader) tries to read a slice of records from KRaft which is greater than KafkaRaftClient#MAX_BATCH_SIZE_BYTES, the RecordsIterator can return records which are past the slice (HWM). As it iterates through this slice of records, FileRecords#readInto reads from disk the minimum of the bytes until end of the file and the remaining space in the associated buffer. In the case of the former, this can lead to a metadata error where the node tries to snapshot with records past the HWM. For the metadata error to occur, the following three pre-conditions must be met: 1. One committed read handed to a listener is larger than 8 MiB. 2. The segment file has bytes after the HWM, meaning records this node already appended but that aren't committed yet. 3. For the error to actually show up, the snapshot generator emits a snapshot from the resulting image before the HWM catches up. The underlying cause is the same as KAFKA-20841. Thanks to @asdf2014 for working on this issue as part of https://github.com/apache/kafka/pull/22963. Change `FileRecords#readInto` which reads from the underlying file to the minimum of: 1. The remaining space in the associated buffer, which in the case of RecordsIterator is the minimum of the FileRecords' record size and the batch size defined by KRaft. 2. The end byte position of the FileRecords object, rather than the end of the underlying file. This ensures when a caller constructs a FileRecords slice of [start position, HWM), no bytes past the HWM are read into memory. Co-authored-by: Benedict Jin <asdf2014@apache.org> Reviewers: José Armando García Sancio <jsancio@apache.org>, Jun Rao <junrao@gmail.com>, Zhiyan Tang <ztang@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>
KRaft (Kafka Raft)
KRaft (Kafka Raft) is a protocol based on the Raft Consensus Protocol tailored for Apache Kafka.
This is used by Apache Kafka in the KRaft (Kafka Raft Metadata) mode. We also have a standalone test server which can be used for performance testing. We describe the details to set this up below.
Run Single Quorum
bin/test-kraft-server-start.sh --config config/kraft.properties --replica-directory-id b8tRS7h4TJ2Vt43Dp85v2A
Run Multi Node Quorum
Create 3 separate KRaft quorum properties as the following:
cat << EOF >> config/kraft-quorum-1.properties
node.id=1
listeners=PLAINTEXT://localhost:9092
controller.listener.names=PLAINTEXT
controller.quorum.voters=1@localhost:9092,2@localhost:9093,3@localhost:9094
log.dirs=/tmp/kraft-logs-1
EOF
cat << EOF >> config/kraft-quorum-2.properties
node.id=2
listeners=PLAINTEXT://localhost:9093
controller.listener.names=PLAINTEXT
controller.quorum.voters=1@localhost:9092,2@localhost:9093,3@localhost:9094
log.dirs=/tmp/kraft-logs-2
EOF
cat << EOF >> config/kraft-quorum-3.properties
node.id=3
listeners=PLAINTEXT://localhost:9094
controller.listener.names=PLAINTEXT
controller.quorum.voters=1@localhost:9092,2@localhost:9093,3@localhost:9094
log.dirs=/tmp/kraft-logs-3
EOF
Open up 3 separate terminals, and run individual commands:
bin/test-kraft-server-start.sh --config config/kraft-quorum-1.properties --replica-directory-id b8tRS7h4TJ2Vt43Dp85v2A
bin/test-kraft-server-start.sh --config config/kraft-quorum-2.properties --replica-directory-id Nkij_D9XRiYKNb41SiJo7Q
bin/test-kraft-server-start.sh --config config/kraft-quorum-3.properties --replica-directory-id 4-e97nI7eHPYKfEDtW8rtQ
Once a leader is elected, it will begin writing to an internal
__raft_performance_test topic with a steady workload of random data.
You can control the workload using the --throughput and --record-size
arguments passed to test-kraft-server-start.sh.