KAFKA-16207; KRaft's internal log listener to update voter set

checkstyle/import-control.xml

@@ -456,12 +456,13 @@
     <allow pkg="org.apache.kafka.snapshot" />
     <allow pkg="org.apache.kafka.clients" />
     <allow pkg="org.apache.kafka.common.config" />
+    <allow pkg="org.apache.kafka.common.feature" />
     <allow pkg="org.apache.kafka.common.message" />
     <allow pkg="org.apache.kafka.common.metadata" />
     <allow pkg="org.apache.kafka.common.metrics" />
+    <allow pkg="org.apache.kafka.common.protocol" />
     <allow pkg="org.apache.kafka.common.record" />
     <allow pkg="org.apache.kafka.common.requests" />
-    <allow pkg="org.apache.kafka.common.protocol" />
     <allow pkg="org.apache.kafka.server.common" />
     <allow pkg="org.apache.kafka.server.common.serialization" />
     <allow pkg="org.apache.kafka.server.fault"/>

clients/src/main/java/org/apache/kafka/clients/admin/SupportedVersionRange.java

@@ -28,7 +28,7 @@ public class SupportedVersionRange {
 
     /**
      * Raises an exception unless the following conditions are met:
-     *  1 <= minVersion <= maxVersion.
+     *  0 <= minVersion <= maxVersion.
      *
      * @param minVersion           The minimum version value.
      * @param maxVersion           The maximum version value.

clients/src/main/java/org/apache/kafka/common/feature/BaseVersionRange.java

@@ -26,7 +26,7 @@
 /**
  * Represents an immutable basic version range using 2 attributes: min and max, each of type short.
  * The min and max attributes need to satisfy 2 rules:
- *  - they are each expected to be >= 1, as we only consider positive version values to be valid.
+ *  - they are each expected to be >= 0, as we only consider positive version values to be valid.
  *  - max should be >= min.
  *
  * The class also provides API to convert the version range to a map.
@@ -48,22 +48,22 @@ class BaseVersionRange {
 
     /**
      * Raises an exception unless the following condition is met:
-     * minValue >= 1 and maxValue >= 1 and maxValue >= minValue.
+     * minValue >= 0 and maxValue >= 0 and maxValue >= minValue.
      *
      * @param minKeyLabel   Label for the min version key, that's used only to convert to/from a map.
      * @param minValue      The minimum version value.
      * @param maxKeyLabel   Label for the max version key, that's used only to convert to/from a map.
      * @param maxValue      The maximum version value.
      *
      * @throws IllegalArgumentException   If any of the following conditions are true:
-     *                                     - (minValue < 1) OR (maxValue < 1) OR (maxValue < minValue).
+     *                                     - (minValue < 0) OR (maxValue < 0) OR (maxValue < minValue).
      *                                     - minKeyLabel is empty, OR, minKeyLabel is empty.
      */
     protected BaseVersionRange(String minKeyLabel, short minValue, String maxKeyLabel, short maxValue) {
-        if (minValue < 1 || maxValue < 1 || maxValue < minValue) {
+        if (minValue < 0 || maxValue < 0 || maxValue < minValue) {
             throw new IllegalArgumentException(
                 String.format(
-                    "Expected minValue >= 1, maxValue >= 1 and maxValue >= minValue, but received" +
+                    "Expected minValue >= 0, maxValue >= 0 and maxValue >= minValue, but received" +
                     " minValue: %d, maxValue: %d", minValue, maxValue));
         }
         if (minKeyLabel.isEmpty()) {
@@ -86,6 +86,7 @@ public short max() {
         return maxValue;
     }
 
+    @Override
     public String toString() {
         return String.format(
             "%s[%s]",

clients/src/main/java/org/apache/kafka/common/feature/SupportedVersionRange.java

@@ -33,7 +33,7 @@ public SupportedVersionRange(short minVersion, short maxVersion) {
     }
 
     public SupportedVersionRange(short maxVersion) {
-        this((short) 1, maxVersion);
+        this((short) 0, maxVersion);
     }
 
     public static SupportedVersionRange fromMap(Map<String, Short> versionRangeMap) {

clients/src/main/java/org/apache/kafka/common/record/ControlRecordType.java

@@ -44,11 +44,15 @@ public enum ControlRecordType {
     ABORT((short) 0),
     COMMIT((short) 1),
 
-    // Raft quorum related control messages.
+    // KRaft quorum related control messages
     LEADER_CHANGE((short) 2),
     SNAPSHOT_HEADER((short) 3),
     SNAPSHOT_FOOTER((short) 4),
 
+    // KRaft membership changes messages
+    KRAFT_VERSION((short) 5),
+    VOTERS((short) 6),
+
     // UNKNOWN is used to indicate a control type which the client is not aware of and should be ignored
     UNKNOWN((short) -1);
 
@@ -108,6 +112,10 @@ public static ControlRecordType fromTypeId(short typeId) {
                 return SNAPSHOT_HEADER;
             case 4:
                 return SNAPSHOT_FOOTER;
+            case 5:
+                return KRAFT_VERSION;
+            case 6:
+                return VOTERS;
 
             default:
                 return UNKNOWN;

clients/src/main/java/org/apache/kafka/common/record/ControlRecordUtils.java

@@ -16,9 +16,11 @@
  */
 package org.apache.kafka.common.record;
 
+import org.apache.kafka.common.message.KRaftVersionRecord;
 import org.apache.kafka.common.message.LeaderChangeMessage;
-import org.apache.kafka.common.message.SnapshotHeaderRecord;
 import org.apache.kafka.common.message.SnapshotFooterRecord;
+import org.apache.kafka.common.message.SnapshotHeaderRecord;
+import org.apache.kafka.common.message.VotersRecord;
 import org.apache.kafka.common.protocol.ByteBufferAccessor;
 
 import java.nio.ByteBuffer;
@@ -27,49 +29,77 @@
  * Utility class for easy interaction with control records.
  */
 public class ControlRecordUtils {
+    public static final short KRAFT_VERSION_CURRENT_VERSION = 0;
     public static final short LEADER_CHANGE_CURRENT_VERSION = 0;
-    public static final short SNAPSHOT_HEADER_CURRENT_VERSION = 0;
     public static final short SNAPSHOT_FOOTER_CURRENT_VERSION = 0;
+    public static final short SNAPSHOT_HEADER_CURRENT_VERSION = 0;
+    public static final short VOTERS_CURRENT_VERSION = 0;
 
     public static LeaderChangeMessage deserializeLeaderChangeMessage(Record record) {
         ControlRecordType recordType = ControlRecordType.parse(record.key());
-        if (recordType != ControlRecordType.LEADER_CHANGE) {
-            throw new IllegalArgumentException(
-                "Expected LEADER_CHANGE control record type(2), but found " + recordType.toString());
-        }
+        validateControlRecordType(ControlRecordType.LEADER_CHANGE, recordType);
+
         return deserializeLeaderChangeMessage(record.value());
     }
 
     public static LeaderChangeMessage deserializeLeaderChangeMessage(ByteBuffer data) {
-        ByteBufferAccessor byteBufferAccessor = new ByteBufferAccessor(data.slice());
-        return new LeaderChangeMessage(byteBufferAccessor, LEADER_CHANGE_CURRENT_VERSION);
+        return new LeaderChangeMessage(new ByteBufferAccessor(data.slice()), LEADER_CHANGE_CURRENT_VERSION);
     }
 
     public static SnapshotHeaderRecord deserializeSnapshotHeaderRecord(Record record) {
         ControlRecordType recordType = ControlRecordType.parse(record.key());
-        if (recordType != ControlRecordType.SNAPSHOT_HEADER) {
-            throw new IllegalArgumentException(
-                "Expected SNAPSHOT_HEADER control record type(3), but found " + recordType.toString());
-        }
+        validateControlRecordType(ControlRecordType.SNAPSHOT_HEADER, recordType);
+
         return deserializeSnapshotHeaderRecord(record.value());
     }
 
     public static SnapshotHeaderRecord deserializeSnapshotHeaderRecord(ByteBuffer data) {
-        ByteBufferAccessor byteBufferAccessor = new ByteBufferAccessor(data.slice());
-        return new SnapshotHeaderRecord(byteBufferAccessor, SNAPSHOT_HEADER_CURRENT_VERSION);
+        return new SnapshotHeaderRecord(new ByteBufferAccessor(data.slice()), SNAPSHOT_HEADER_CURRENT_VERSION);
     }
 
     public static SnapshotFooterRecord deserializeSnapshotFooterRecord(Record record) {
         ControlRecordType recordType = ControlRecordType.parse(record.key());
-        if (recordType != ControlRecordType.SNAPSHOT_FOOTER) {
-            throw new IllegalArgumentException(
-                "Expected SNAPSHOT_FOOTER control record type(4), but found " + recordType.toString());
-        }
+        validateControlRecordType(ControlRecordType.SNAPSHOT_FOOTER, recordType);
+
         return deserializeSnapshotFooterRecord(record.value());
     }
 
     public static SnapshotFooterRecord deserializeSnapshotFooterRecord(ByteBuffer data) {
-        ByteBufferAccessor byteBufferAccessor = new ByteBufferAccessor(data.slice());
-        return new SnapshotFooterRecord(byteBufferAccessor, SNAPSHOT_FOOTER_CURRENT_VERSION);
+        return new SnapshotFooterRecord(new ByteBufferAccessor(data.slice()), SNAPSHOT_FOOTER_CURRENT_VERSION);
+    }
+
+    public static KRaftVersionRecord deserializeKRaftVersionRecord(Record record) {
+        ControlRecordType recordType = ControlRecordType.parse(record.key());
+        validateControlRecordType(ControlRecordType.KRAFT_VERSION, recordType);
+
+        return deserializeKRaftVersionRecord(record.value());
+    }
+
+    public static KRaftVersionRecord deserializeKRaftVersionRecord(ByteBuffer data) {
+        return new KRaftVersionRecord(new ByteBufferAccessor(data.slice()), KRAFT_VERSION_CURRENT_VERSION);
+    }
+
+    public static VotersRecord deserializeVotersRecord(Record record) {
+        ControlRecordType recordType = ControlRecordType.parse(record.key());
+        validateControlRecordType(ControlRecordType.VOTERS, recordType);
+
+        return deserializeVotersRecord(record.value());
+    }
+
+    public static VotersRecord deserializeVotersRecord(ByteBuffer data) {
+        return new VotersRecord(new ByteBufferAccessor(data.slice()), VOTERS_CURRENT_VERSION);
+    }
+
+    private static void validateControlRecordType(ControlRecordType expected, ControlRecordType actual) {
+        if (actual != expected) {
+            throw new IllegalArgumentException(
+                String.format(
+                    "Expected %s control record type(%d), but found %s",
+                    expected,
+                    expected.type(),
+                    actual
+                )
+            );
+        }
     }
 }

clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java

@@ -18,9 +18,11 @@
 
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.CorruptRecordException;
+import org.apache.kafka.common.message.KRaftVersionRecord;
 import org.apache.kafka.common.message.LeaderChangeMessage;
-import org.apache.kafka.common.message.SnapshotHeaderRecord;
 import org.apache.kafka.common.message.SnapshotFooterRecord;
+import org.apache.kafka.common.message.SnapshotHeaderRecord;
+import org.apache.kafka.common.message.VotersRecord;
 import org.apache.kafka.common.network.TransferableChannel;
 import org.apache.kafka.common.record.MemoryRecords.RecordFilter.BatchRetention;
 import org.apache.kafka.common.record.MemoryRecords.RecordFilter.BatchRetentionResult;
@@ -154,7 +156,7 @@ public FilterResult filterTo(TopicPartition partition, RecordFilter filter, Byte
 
     /**
      * Note: This method is also used to convert the first timestamp of the batch (which is usually the timestamp of the first record)
-     * to the delete horizon of the tombstones or txn markers which are present in the batch. 
+     * to the delete horizon of the tombstones or txn markers which are present in the batch.
      */
     private static FilterResult filterTo(TopicPartition partition, Iterable<MutableRecordBatch> batches,
                                          RecordFilter filter, ByteBuffer destinationBuffer, int maxRecordBatchSize,
@@ -807,4 +809,62 @@ private static void writeSnapshotFooterRecord(
             builder.appendSnapshotFooterMessage(timestamp, snapshotFooterRecord);
         }
     }
+
+    public static MemoryRecords withKRaftVersionRecord(
+        long initialOffset,
+        long timestamp,
+        int leaderEpoch,
+        ByteBuffer buffer,
+        KRaftVersionRecord kraftVersionRecord
+    ) {
+        writeKRaftVersionRecord(buffer, initialOffset, timestamp, leaderEpoch, kraftVersionRecord);
+        buffer.flip();
+        return MemoryRecords.readableRecords(buffer);
+    }
+
+    private static void writeKRaftVersionRecord(
+        ByteBuffer buffer,
+        long initialOffset,
+        long timestamp,
+        int leaderEpoch,
+        KRaftVersionRecord kraftVersionRecord
+    ) {
+        try (MemoryRecordsBuilder builder = new MemoryRecordsBuilder(
+            buffer, RecordBatch.CURRENT_MAGIC_VALUE, CompressionType.NONE,
+            TimestampType.CREATE_TIME, initialOffset, timestamp,
+            RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE,
+            false, true, leaderEpoch, buffer.capacity())
+        ) {
+            builder.appendKRaftVersionMessage(timestamp, kraftVersionRecord);
+        }
+    }
+
+    public static MemoryRecords withVotersRecord(
+        long initialOffset,
+        long timestamp,
+        int leaderEpoch,
+        ByteBuffer buffer,
+        VotersRecord votersRecord
+    ) {
+        writeVotersRecord(buffer, initialOffset, timestamp, leaderEpoch, votersRecord);
+        buffer.flip();
+        return MemoryRecords.readableRecords(buffer);
+    }
+
+    private static void writeVotersRecord(
+        ByteBuffer buffer,
+        long initialOffset,
+        long timestamp,
+        int leaderEpoch,
+        VotersRecord votersRecord
+    ) {
+        try (MemoryRecordsBuilder builder = new MemoryRecordsBuilder(
+            buffer, RecordBatch.CURRENT_MAGIC_VALUE, CompressionType.NONE,
+            TimestampType.CREATE_TIME, initialOffset, timestamp,
+            RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE,
+            false, true, leaderEpoch, buffer.capacity())
+        ) {
+            builder.appendVotersMessage(timestamp, votersRecord);
+        }
+    }
 }

clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java

@@ -18,9 +18,11 @@
 
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.message.KRaftVersionRecord;
 import org.apache.kafka.common.message.LeaderChangeMessage;
-import org.apache.kafka.common.message.SnapshotHeaderRecord;
 import org.apache.kafka.common.message.SnapshotFooterRecord;
+import org.apache.kafka.common.message.SnapshotHeaderRecord;
+import org.apache.kafka.common.message.VotersRecord;
 import org.apache.kafka.common.protocol.MessageUtil;
 import org.apache.kafka.common.protocol.types.Struct;
 import org.apache.kafka.common.utils.ByteBufferOutputStream;
@@ -602,11 +604,12 @@ public void append(SimpleRecord record) {
 
     /**
      * Append a control record at the next sequential offset.
+     *
      * @param timestamp The record timestamp
      * @param type The control record type (cannot be UNKNOWN)
      * @param value The control record value
      */
-    private void appendControlRecord(long timestamp, ControlRecordType type, ByteBuffer value) {
+    public void appendControlRecord(long timestamp, ControlRecordType type, ByteBuffer value) {
         Struct keyStruct = type.recordKey();
         ByteBuffer key = ByteBuffer.allocate(keyStruct.sizeOf());
         keyStruct.writeTo(key);
@@ -650,6 +653,22 @@ public void appendSnapshotFooterMessage(long timestamp, SnapshotFooterRecord sna
         );
     }
 
+    public void appendKRaftVersionMessage(long timestamp, KRaftVersionRecord kraftVersionRecord) {
+        appendControlRecord(
+            timestamp,
+            ControlRecordType.KRAFT_VERSION,
+            MessageUtil.toByteBuffer(kraftVersionRecord, ControlRecordUtils.KRAFT_VERSION_CURRENT_VERSION)
+        );
+    }
+
+    public void appendVotersMessage(long timestamp, VotersRecord votersRecord) {
+        appendControlRecord(
+            timestamp,
+            ControlRecordType.VOTERS,
+            MessageUtil.toByteBuffer(votersRecord, ControlRecordUtils.VOTERS_CURRENT_VERSION)
+        );
+    }
+
     /**
      * Add a legacy record without doing offset/magic validation (this should only be used in testing).
      * @param offset The offset of the record

clients/src/main/resources/common/message/KRaftVersionRecord.json

@@ -0,0 +1,27 @@
+// 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.
+
+{
+  "type": "data",
+  "name": "KRaftVersionRecord",
+  "validVersions": "0",
+  "flexibleVersions": "0+",
+  "fields": [
+    { "name": "Version", "type": "int16", "versions": "0+",
+      "about": "The version of the kraft version record" },
+    { "name": "KRaftVersion", "type": "int16", "versions": "0+",
+      "about": "The kraft protocol version" }
+  ]
+}