This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 8935db691dd KAFKA-20659: Reject unsupported MirrorMaker record
versions at runtime (#22504)
8935db691dd is described below
commit 8935db691dd927d6352b51df6365f4787eae7642
Author: nileshkumar3 <[email protected]>
AuthorDate: Wed Jun 24 13:16:32 2026 -0500
KAFKA-20659: Reject unsupported MirrorMaker record versions at runtime
(#22504)
Reviewers: Mickael Maison <[email protected]>
---
.../apache/kafka/connect/mirror/Checkpoint.java | 5 +-
.../org/apache/kafka/connect/mirror/Heartbeat.java | 5 +-
.../connect/mirror/MirrorRecordVersionTest.java | 88 ++++++++++++++++++++++
3 files changed, 96 insertions(+), 2 deletions(-)
diff --git
a/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Checkpoint.java
b/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Checkpoint.java
index 3e0a2ee6177..3bfbca81bbc 100644
---
a/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Checkpoint.java
+++
b/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Checkpoint.java
@@ -19,6 +19,7 @@ package org.apache.kafka.connect.mirror;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
import org.apache.kafka.common.protocol.types.Struct;
@@ -138,7 +139,9 @@ public class Checkpoint {
}
private static Schema valueSchema(short version) {
- assert version == 0;
+ if (version != VERSION) {
+ throw new UnsupportedVersionException("Unsupported version " +
version);
+ }
return VALUE_SCHEMA_V0;
}
diff --git
a/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Heartbeat.java
b/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Heartbeat.java
index d63dfa70ff8..42bc666fdd7 100644
---
a/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Heartbeat.java
+++
b/connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/Heartbeat.java
@@ -17,6 +17,7 @@
package org.apache.kafka.connect.mirror;
import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
import org.apache.kafka.common.protocol.types.Struct;
@@ -140,7 +141,9 @@ public class Heartbeat {
}
private static Schema valueSchema(short version) {
- assert version == 0;
+ if (version != VERSION) {
+ throw new UnsupportedVersionException("Unsupported version " +
version);
+ }
return VALUE_SCHEMA_V0;
}
}
diff --git
a/connect/mirror-client/src/test/java/org/apache/kafka/connect/mirror/MirrorRecordVersionTest.java
b/connect/mirror-client/src/test/java/org/apache/kafka/connect/mirror/MirrorRecordVersionTest.java
new file mode 100644
index 00000000000..628975d918f
--- /dev/null
+++
b/connect/mirror-client/src/test/java/org/apache/kafka/connect/mirror/MirrorRecordVersionTest.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+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;
+ }
+}
