feat(failover): add wal failover support (#2516) (#2517)

Signed-off-by: Robin Han <[email protected]>

Author: superhx

config/kraft/broker.properties

@@ -143,8 +143,8 @@ s3.data.buckets=0@s3://ko3?region=us-east-1
 # The ops buckets
 s3.ops.buckets=0@s3://ko3?region=us-east-1
 
-# The file path of delta WAL in block device
-s3.wal.path=0@file:///tmp/kraft-broker-logs/s3wal?capacity=2147483648
+# The wal storage config
+s3.wal.path=0@s3://ko3?region=us-east-1
 
 # The maximum size of WAL cache can use, default 2GB
 # s3.wal.cache.size=2147483648

config/kraft/server.properties

@@ -146,8 +146,8 @@ s3.data.buckets=0@s3://ko3?region=us-east-1
 # The ops buckets
 s3.ops.buckets=0@s3://ko3?region=us-east-1
 
-# The file path of delta WAL in block device
-s3.wal.path=0@file:///tmp/kraft-broker-logs/s3wal?capacity=2147483648
+# The wal storage config
+s3.wal.path=0@s3://ko3?region=us-east-1
 
 # The maximum size of WAL cache can use, default 2GB
 # s3.wal.cache.size=2147483648

core/src/main/java/kafka/automq/AutoMQConfig.java

@@ -56,8 +56,7 @@ public class AutoMQConfig {
     public static final String ELASTIC_STREAM_ENABLE_DOC = "Whether to enable AutoMQ, it has to be set to true";
 
     public static final String ELASTIC_STREAM_ENDPOINT_CONFIG = "elasticstream.endpoint";
-    public static final String ELASTIC_STREAM_ENDPOINT_DOC = "Specifies the Elastic Stream endpoint, ex. <code>es://hostname1:port1,hostname2:port2,hostname3:port3</code>.\n" +
-        "You could also PoC launch it in memory mode with endpoint <code>memory:://</code> or redis mode with <code>redis://.</code>";
+    public static final String ELASTIC_STREAM_ENDPOINT_DOC = "Specifies the Elastic Stream endpoint";
 
     public static final String S3_DATA_BUCKETS_CONFIG = "s3.data.buckets";
     public static final String S3_DATA_BUCKETS_DOC = "The data buckets url with format 0@s3://$bucket?region=$region. \n" +
@@ -69,8 +68,7 @@ public class AutoMQConfig {
     public static final String S3_OPS_BUCKETS_DOC = "With the same format as s3.data.buckets";
 
     public static final String S3_WAL_PATH_CONFIG = "s3.wal.path";
-    public static final String S3_WAL_PATH_DOC = "The local WAL path for AutoMQ can be set to a block device path such as 0@file:///dev/xxx?iops=3000&iodepth=8&iobandwidth=157286400 or a filesystem file path." +
-        "It is recommended to use a block device for better write performance.";
+    public static final String S3_WAL_PATH_DOC = "The WAL path for AutoMQ, The format is '0@s3://$bucket?region=$region[&batchInterval=250][&maxBytesInBatch=8388608]'";
 
     public static final String S3_WAL_CACHE_SIZE_CONFIG = "s3.wal.cache.size";
     public static final String S3_WAL_CACHE_SIZE_DOC = "The WAL (Write-Ahead Log) cache is a FIFO (First In, First Out) queue that contains data that has not yet been uploaded to object storage, as well as data that has been uploaded but not yet evicted from the cache." +

core/src/main/java/kafka/automq/controller/DefaultQuorumControllerExtension.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2025, AutoMQ HK Limited.
+ *
+ * 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 kafka.automq.controller;
+
+import kafka.automq.failover.FailoverControlManager;
+
+import org.apache.kafka.common.metadata.KVRecord;
+import org.apache.kafka.common.metadata.MetadataRecordType;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.controller.QuorumController;
+import org.apache.kafka.controller.QuorumControllerExtension;
+import org.apache.kafka.raft.OffsetAndEpoch;
+
+import java.util.Objects;
+import java.util.Optional;
+
+public class DefaultQuorumControllerExtension implements QuorumControllerExtension {
+    private final FailoverControlManager failoverControlManager;
+
+    public DefaultQuorumControllerExtension(QuorumController controller) {
+        this.failoverControlManager = new FailoverControlManager(
+            controller.snapshotRegistry(),
+            controller,
+            controller.clusterControl(),
+            controller.nodeControlManager(),
+            controller.streamControlManager()
+        );
+    }
+
+    @Override
+    public boolean replay(MetadataRecordType type, ApiMessage message, Optional<OffsetAndEpoch> snapshotId,
+        long batchLastOffset) {
+        if (Objects.requireNonNull(type) == MetadataRecordType.KVRECORD) {
+            failoverControlManager.replay((KVRecord) message);
+        } else {
+            return false;
+        }
+        return true;
+    }
+}

core/src/main/java/kafka/automq/failover/DefaultFailedWal.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2025, AutoMQ HK Limited.
+ *
+ * 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 kafka.automq.failover;
+
+import java.util.Objects;
+
+public final class DefaultFailedWal implements FailedWal {
+    private final NodeRuntimeMetadata nodeMetadata;
+
+    public DefaultFailedWal(NodeRuntimeMetadata nodeMetadata) {
+        this.nodeMetadata = nodeMetadata;
+    }
+
+    public NodeRuntimeMetadata nodeMetadata() {
+        return nodeMetadata;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == this)
+            return true;
+        if (obj == null || obj.getClass() != this.getClass())
+            return false;
+        var that = (DefaultFailedWal) obj;
+        return Objects.equals(this.nodeMetadata, that.nodeMetadata);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(nodeMetadata);
+    }
+
+    @Override
+    public String toString() {
+        return "FailedWalV1[" +
+            "nodeMetadata=" + nodeMetadata + ']';
+    }
+
+    @Override
+    public FailoverContext toFailoverContext(int target) {
+        return new FailoverContext(nodeMetadata.id(), nodeMetadata.epoch(), target, nodeMetadata.walConfigs());
+    }
+
+    @Override
+    public FailedNode node() {
+        return FailedNode.from(nodeMetadata);
+    }
+
+    public static FailedWal from(NodeRuntimeMetadata failedNode) {
+        return new DefaultFailedWal(failedNode);
+    }
+}

core/src/main/java/kafka/automq/failover/FailedNode.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2025, AutoMQ HK Limited.
+ *
+ * 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 kafka.automq.failover;
+
+public interface FailedNode {
+
+    /**
+     * The node id of the failed node.
+     */
+    int id();
+
+    static FailedNode from(NodeRuntimeMetadata node) {
+        return new K8sFailedNode(node.id());
+    }
+
+    static FailedNode from(FailoverContext context) {
+        return new K8sFailedNode(context.getNodeId());
+    }
+}

core/src/main/java/kafka/automq/failover/FailedWal.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2025, AutoMQ HK Limited.
+ *
+ * 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 kafka.automq.failover;
+
+public interface FailedWal {
+
+    /**
+     * Convert to a failover context.
+     */
+    FailoverContext toFailoverContext(int target);
+
+    /**
+     * The node of the failed WAL belongs to.
+     */
+    FailedNode node();
+
+    default int nodeId() {
+        return node().id();
+    }
+}

core/src/main/java/kafka/automq/failover/FailoverConstants.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2025, AutoMQ HK Limited.
+ *
+ * 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 kafka.automq.failover;
+
+public class FailoverConstants {
+    public static final String FAILOVER_KEY = "__a.failover";
+
+}

core/src/main/java/kafka/automq/failover/FailoverContext.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2025, AutoMQ HK Limited.
+ *
+ * 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 kafka.automq.failover;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class FailoverContext {
+    /**
+     * Failed node id
+     */
+    @JsonProperty("n")
+    private int nodeId;
+
+    /**
+     * Failover target node id
+     *
+     * @since failover v0
+     */
+    @JsonProperty("t")
+    private int target;
+
+    /**
+     * Failed node epoch
+     *
+     * @since failover v1
+     */
+    @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+    @JsonProperty(value = "e", defaultValue = "0")
+    private long nodeEpoch;
+
+    /**
+     * WAL configs for failover
+     */
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @JsonProperty("c")
+    private String kraftWalConfigs;
+
+    // for json deserialize
+    public FailoverContext() {}
+
+    public FailoverContext(int nodeId, long nodeEpoch, int target, String kraftWalConfigs) {
+        this.nodeId = nodeId;
+        this.nodeEpoch = nodeEpoch;
+        this.target = target;
+        this.kraftWalConfigs = kraftWalConfigs;
+    }
+
+    @JsonIgnore
+    public FailedNode getFailedNode() {
+        return FailedNode.from(this);
+    }
+
+    public int getNodeId() {
+        return nodeId;
+    }
+
+    public int getTarget() {
+        return target;
+    }
+
+    public long getNodeEpoch() {
+        return nodeEpoch;
+    }
+
+    public String getKraftWalConfigs() {
+        return kraftWalConfigs;
+    }
+
+    @Override
+    public String toString() {
+        return "FailoverContext{" +
+            "nodeId=" + nodeId +
+            ", target=" + target +
+            ", nodeEpoch=" + nodeEpoch +
+            ", kraftWalConfigs=" + kraftWalConfigs +
+            '}';
+    }
+}