[jira] [Assigned] (KAFKA-20659) MirrorMaker Checkpoint and Heartbeat records can accept unsupported versions when assertions are disabled

["Mickael Maison (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/KAFKA-20659?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Mickael Maison reassigned KAFKA-20659:
--------------------------------------

    Assignee: Nilesh Kumar

> 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
>
> *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)