[
https://issues.apache.org/jira/browse/KAFKA-20659?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Mickael Maison resolved KAFKA-20659.
------------------------------------
Fix Version/s: 4.4.0
Resolution: Fixed
> MirrorMaker Checkpoint and Heartbeat records can accept unsupported versions
> when assertions are disabled
> ---------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-20659
> URL: https://issues.apache.org/jira/browse/KAFKA-20659
> Project: Kafka
> Issue Type: Bug
> Components: connect
> Reporter: Ruiqi Dong
> Assignee: Nilesh Kumar
> Priority: Major
> Fix For: 4.4.0
>
>
> *Summary*
> `Checkpoint.deserializeRecord(...)` and `Heartbeat.deserializeRecord(...)`
> read a version field from the record header, but the version check is
> implemented only as:
> {code:java}
> assert version == 0; {code}
> This is not a reliable runtime validation mechanism. In normal production
> JVMs, assertions are disabled by default, so the check disappears, and an
> unknown future version is parsed with the v0 schema. In assertion-enabled
> test runs, the same input fails with `AssertionError`, which is still not an
> appropriate public failure mode for malformed record data.
> *Affected code*
> File:
> connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Checkpoint.java
> {code:java}
> private static Schema valueSchema(short version) {
> assert version == 0;
> return VALUE_SCHEMA_V0;
> } {code}
> File:
> connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Heartbeat.java
> {code:java}
> private static Schema valueSchema(short version) {
> assert version == 0;
> return VALUE_SCHEMA_V0;
> } {code}
> *Reproducer*
> Add this test to
> connect/mirror-client/src/test/java/org/apache/kafka/connect/mirror/MirrorRecordVersionTest.java:
> The test expects an explicit Kafka exception. In assertion-enabled test runs,
> it currently observes `AssertionError`; in normal `-da` JVMs, the `assert` is
> skipped and the unsupported version is interpreted as v0.
> {code:java}
> package org.apache.kafka.connect.mirror;
> import org.apache.kafka.clients.consumer.ConsumerRecord;
> import org.apache.kafka.common.TopicPartition;
> import org.apache.kafka.common.errors.UnsupportedVersionException;
> import org.apache.kafka.common.protocol.types.Schema;
> import org.apache.kafka.common.protocol.types.Struct;
> import org.junit.jupiter.api.Test;
> import java.nio.ByteBuffer;
> import static org.junit.jupiter.api.Assertions.assertThrows;
> public class MirrorRecordVersionTest {
> @Test
> public void checkpointDeserializeRejectsUnsupportedValueVersion() {
> Checkpoint checkpoint = new Checkpoint("group", new
> TopicPartition("topic", 0), 10L, 20L, "metadata");
> ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>(
> "checkpoints",
> 0,
> 0,
> checkpoint.recordKey(),
> checkpointValueWithVersion((short) (Checkpoint.VERSION + 1))
> );
> assertThrows(UnsupportedVersionException.class, () ->
> Checkpoint.deserializeRecord(record));
> }
> @Test
> public void heartbeatDeserializeRejectsUnsupportedValueVersion() {
> Heartbeat heartbeat = new Heartbeat("source", "target", 123L);
> ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>(
> "heartbeats",
> 0,
> 0,
> heartbeat.recordKey(),
> heartbeatValueWithVersion((short) (Heartbeat.VERSION + 1))
> );
> assertThrows(UnsupportedVersionException.class, () ->
> Heartbeat.deserializeRecord(record));
> }
> private static byte[] checkpointValueWithVersion(short version) {
> Struct header = new Struct(Checkpoint.HEADER_SCHEMA);
> header.set(Checkpoint.VERSION_KEY, version);
> Struct value = new Struct(Checkpoint.VALUE_SCHEMA_V0);
> value.set(Checkpoint.UPSTREAM_OFFSET_KEY, 10L);
> value.set(Checkpoint.DOWNSTREAM_OFFSET_KEY, 20L);
> value.set(Checkpoint.METADATA_KEY, "metadata");
> return write(Checkpoint.HEADER_SCHEMA, header,
> Checkpoint.VALUE_SCHEMA_V0, value);
> }
> private static byte[] heartbeatValueWithVersion(short version) {
> Struct header = new Struct(Heartbeat.HEADER_SCHEMA);
> header.set(Heartbeat.VERSION_KEY, version);
> Struct value = new Struct(Heartbeat.VALUE_SCHEMA_V0);
> value.set(Heartbeat.TIMESTAMP_KEY, 123L);
> return write(Heartbeat.HEADER_SCHEMA, header,
> Heartbeat.VALUE_SCHEMA_V0, value);
> }
> private static byte[] write(Schema headerSchema, Struct header, Schema
> valueSchema, Struct value) {
> ByteBuffer buffer = ByteBuffer.allocate(headerSchema.sizeOf(header) +
> valueSchema.sizeOf(value));
> headerSchema.write(buffer, header);
> valueSchema.write(buffer, value);
> buffer.flip();
> byte[] result = new byte[buffer.remaining()];
> buffer.get(result);
> return result;
> }
> } {code}
> Run:
> {code:java}
> ./gradlew -q :connect:mirror-client:test --tests MirrorRecordVersionTest
> {code}
> Observed behavior:
> With assertions enabled, both tests fail. The stack traces point to
> `Checkpoint.valueSchema(...)` and `Heartbeat.valueSchema(...)`. With
> assertions disabled, the same version check is not executed, so the record is
> parsed using `VALUE_SCHEMA_V0`.
> {code:java}
> Unexpected exception type thrown,
> expected: UnsupportedVersionException
> but was: AssertionError {code}
> Expected behavior:
> Unsupported record versions should be rejected deterministically, regardless
> of the JVM assertion setting, with an explicit runtime exception such as
> `UnsupportedVersionException`.
> MirrorMaker checkpoint records determine translated consumer offsets, and
> heartbeat records determine replication path/liveness. Silently interpreting
> a future or corrupt record version as v0 can misread replication metadata.
> The `AssertionError` observed under `-ea` is a secondary symptom; the main
> issue is that the runtime validation is absent under the default `-da`
> setting.
> The fix direction is to replace the `assert` with an explicit runtime check:
> {code:java}
> if (version != VERSION) {
> throw new UnsupportedVersionException("Unsupported version " + version);
> } {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
