fix: add kafka-go metadata compatibility mode

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

@@ -43,6 +43,7 @@
 import static org.apache.kafka.common.config.ConfigDef.Importance.HIGH;
 import static org.apache.kafka.common.config.ConfigDef.Importance.LOW;
 import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
+import static org.apache.kafka.common.config.ConfigDef.Range.atLeast;
 import static org.apache.kafka.common.config.ConfigDef.Range.between;
 import static org.apache.kafka.common.config.ConfigDef.Type.BOOLEAN;
 import static org.apache.kafka.common.config.ConfigDef.Type.INT;
@@ -202,6 +203,12 @@ public class AutoMQConfig {
     public static final String RETRY_STORM_BACKOFF_ENABLED_DOC = "Whether retry storm delayed response backoff is enabled";
     public static final boolean RETRY_STORM_BACKOFF_ENABLED_DEFAULT = false;
 
+    public static final String KAFKA_GO_METADATA_COMPATIBILITY_ENABLED_CONFIG =
+        "automq.kafka-go.metadata.compatibility.enabled";
+    public static final String KAFKA_GO_METADATA_COMPATIBILITY_ENABLED_DOC =
+        "Whether kafka-go Metadata compatibility mode is enabled";
+    public static final boolean KAFKA_GO_METADATA_COMPATIBILITY_ENABLED_DEFAULT = true;
+
     public static final String RETRY_STORM_BACKOFF_MAX_DELAY_MS_CONFIG = "automq.retry.storm.backoff.max.delay.ms";
     public static final String RETRY_STORM_BACKOFF_MAX_DELAY_MS_DOC = "The maximum retry storm delayed response time in milliseconds, from 0 to 10000";
     public static final long RETRY_STORM_BACKOFF_MAX_DELAY_MS_DEFAULT = 1000L;
@@ -310,9 +317,10 @@ public static void define(ConfigDef configDef) {
             .define(AutoMQConfig.S3_TELEMETRY_METRICS_EXPORTER_URI_CONFIG, PASSWORD, null, HIGH, AutoMQConfig.S3_TELEMETRY_METRICS_EXPORTER_URI_DOC)
             .define(AutoMQConfig.S3_TELEMETRY_METRICS_BASE_LABELS_CONFIG, STRING, null, MEDIUM, AutoMQConfig.S3_TELEMETRY_METRICS_BASE_LABELS_DOC)
             .define(AutoMQConfig.S3_BACK_PRESSURE_ENABLED_CONFIG, BOOLEAN, AutoMQConfig.S3_BACK_PRESSURE_ENABLED_DEFAULT, MEDIUM, AutoMQConfig.S3_BACK_PRESSURE_ENABLED_DOC)
-            .define(AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_CONFIG, LONG, AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_DEFAULT, MEDIUM, AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_DOC)
+            .define(AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_CONFIG, LONG, AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_DEFAULT, atLeast(0), MEDIUM, AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_DOC)
             .define(AutoMQConfig.RETRY_STORM_BACKOFF_ENABLED_CONFIG, BOOLEAN, AutoMQConfig.RETRY_STORM_BACKOFF_ENABLED_DEFAULT, MEDIUM, AutoMQConfig.RETRY_STORM_BACKOFF_ENABLED_DOC)
             .define(AutoMQConfig.RETRY_STORM_BACKOFF_MAX_DELAY_MS_CONFIG, LONG, AutoMQConfig.RETRY_STORM_BACKOFF_MAX_DELAY_MS_DEFAULT, between(0, AutoMQConfig.RETRY_STORM_BACKOFF_MAX_DELAY_MS_MAX), MEDIUM, AutoMQConfig.RETRY_STORM_BACKOFF_MAX_DELAY_MS_DOC)
+            .define(AutoMQConfig.KAFKA_GO_METADATA_COMPATIBILITY_ENABLED_CONFIG, BOOLEAN, AutoMQConfig.KAFKA_GO_METADATA_COMPATIBILITY_ENABLED_DEFAULT, MEDIUM, AutoMQConfig.KAFKA_GO_METADATA_COMPATIBILITY_ENABLED_DOC)
             .define(AutoMQConfig.ZONE_ROUTER_CHANNELS_CONFIG, ConfigDef.Type.STRING, null, ConfigDef.Importance.HIGH, AutoMQConfig.ZONE_ROUTER_CHANNELS_DOC)
             // Deprecated config start
             .define(AutoMQConfig.S3_ENDPOINT_CONFIG, STRING, null, HIGH, AutoMQConfig.S3_ENDPOINT_DOC)

core/src/main/java/kafka/automq/backpressure/BackPressureConfig.java

@@ -59,29 +59,18 @@ public BackPressureConfig(boolean enabled, long cooldownMs) {
         this.cooldownMs = cooldownMs;
     }
 
-    public static void validate(Map<String, ?> raw) throws ConfigException {
-        Map<String, Object> configs = new HashMap<>(raw);
-        if (configs.containsKey(AutoMQConfig.S3_BACK_PRESSURE_ENABLED_CONFIG)) {
-            ConfigUtils.getBoolean(configs, AutoMQConfig.S3_BACK_PRESSURE_ENABLED_CONFIG);
-        }
-        if (configs.containsKey(AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_CONFIG)) {
-            validateCooldownMs(ConfigUtils.getLong(configs, AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_CONFIG));
-        }
-    }
-
     public static void validateCooldownMs(long cooldownMs) throws ConfigException {
         if (cooldownMs < 0) {
             throw new ConfigException(AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_CONFIG, cooldownMs, "The cooldown time must be non-negative.");
         }
     }
 
     public void update(Map<String, ?> raw) {
-        Map<String, Object> configs = new HashMap<>(raw);
-        if (configs.containsKey(AutoMQConfig.S3_BACK_PRESSURE_ENABLED_CONFIG)) {
-            this.enabled = ConfigUtils.getBoolean(configs, AutoMQConfig.S3_BACK_PRESSURE_ENABLED_CONFIG);
+        if (raw.containsKey(AutoMQConfig.S3_BACK_PRESSURE_ENABLED_CONFIG)) {
+            this.enabled = (Boolean) raw.get(AutoMQConfig.S3_BACK_PRESSURE_ENABLED_CONFIG);
         }
-        if (configs.containsKey(AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_CONFIG)) {
-            this.cooldownMs = ConfigUtils.getLong(configs, AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_CONFIG);
+        if (raw.containsKey(AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_CONFIG)) {
+            this.cooldownMs = (Long) raw.get(AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_CONFIG);
         }
     }
 

core/src/main/java/kafka/automq/backpressure/DefaultBackPressureManager.java

@@ -181,7 +181,6 @@ public Set<String> reconfigurableConfigs() {
 
     @Override
     public void validateReconfiguration(Map<String, ?> configs) throws ConfigException {
-        BackPressureConfig.validate(configs);
     }
 
     @Override

core/src/main/java/kafka/automq/metadata/KafkaGoMetadataResponseRewriter.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2026, 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.metadata;
+
+import kafka.automq.AutoMQConfig;
+import kafka.server.KafkaConfig;
+
+import org.apache.kafka.common.Reconfigurable;
+import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopic;
+import org.apache.kafka.common.protocol.Errors;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Rewrites unavailable kafka-go partition metadata to route requests to the controller-selected first replica.
+ */
+public final class KafkaGoMetadataResponseRewriter implements Reconfigurable {
+
+    public static final Set<String> RECONFIGURABLE_CONFIGS =
+        Set.of(AutoMQConfig.KAFKA_GO_METADATA_COMPATIBILITY_ENABLED_CONFIG);
+
+    private static final String KAFKA_GO_CLIENT_ID = "github.com/segmentio/kafka-go";
+    private volatile boolean enabled;
+
+    public KafkaGoMetadataResponseRewriter(KafkaConfig config) {
+        this.enabled = config.kafkaGoMetadataCompatibilityEnabled();
+    }
+
+    @Override
+    public Set<String> reconfigurableConfigs() {
+        return RECONFIGURABLE_CONFIGS;
+    }
+
+    @Override
+    public void validateReconfiguration(Map<String, ?> configs) {
+    }
+
+    @Override
+    public void reconfigure(Map<String, ?> configs) {
+        if (configs.containsKey(AutoMQConfig.KAFKA_GO_METADATA_COMPATIBILITY_ENABLED_CONFIG)) {
+            enabled = (Boolean) configs.get(AutoMQConfig.KAFKA_GO_METADATA_COMPATIBILITY_ENABLED_CONFIG);
+        }
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs) {
+    }
+
+    /**
+     * Rewrites each leader-unavailable partition with replicas for clients using kafka-go's default client ID.
+     * All partition fields other than the error code and leader ID are preserved.
+     *
+     * @param clientId request header client ID
+     * @param topics mutable Metadata response topics
+     */
+    public void rewrite(String clientId, Collection<MetadataResponseTopic> topics) {
+        if (!enabled
+            || clientId == null
+            || (!clientId.isEmpty() && !clientId.contains(KAFKA_GO_CLIENT_ID))) {
+            return;
+        }
+        topics.forEach(topic -> topic.partitions().forEach(partition -> {
+            if (partition.errorCode() == Errors.LEADER_NOT_AVAILABLE.code()
+                && !partition.replicaNodes().isEmpty()) {
+                partition.setErrorCode(Errors.NONE.code());
+                partition.setLeaderId(partition.replicaNodes().get(0));
+            }
+        }));
+    }
+}

core/src/main/java/kafka/automq/retrystorm/RetryStormBackoffConfig.java

@@ -76,19 +76,6 @@ public static RetryStormBackoffConfig from(Map<String, ?> raw) {
         );
     }
 
-    /**
-     * Validates retry storm keys present in a dynamic broker config update.
-     */
-    public static void validate(Map<String, ?> raw) throws ConfigException {
-        Map<String, Object> configs = new HashMap<>(raw);
-        if (configs.containsKey(AutoMQConfig.RETRY_STORM_BACKOFF_ENABLED_CONFIG)) {
-            ConfigUtils.getBoolean(configs, AutoMQConfig.RETRY_STORM_BACKOFF_ENABLED_CONFIG);
-        }
-        if (configs.containsKey(AutoMQConfig.RETRY_STORM_BACKOFF_MAX_DELAY_MS_CONFIG)) {
-            validateMaxDelayMs(ConfigUtils.getLong(configs, AutoMQConfig.RETRY_STORM_BACKOFF_MAX_DELAY_MS_CONFIG));
-        }
-    }
-
     /**
      * Rejects max delay values outside the bounded delayed-response range.
      */
@@ -104,14 +91,11 @@ public static void validateMaxDelayMs(long maxDelayMs) throws ConfigException {
      * Applies a partial dynamic update; keys absent from {@code raw} keep their current runtime values.
      */
     public void update(Map<String, ?> raw) {
-        Map<String, Object> configs = new HashMap<>(raw);
-        if (configs.containsKey(AutoMQConfig.RETRY_STORM_BACKOFF_ENABLED_CONFIG)) {
-            this.enabled = ConfigUtils.getBoolean(configs, AutoMQConfig.RETRY_STORM_BACKOFF_ENABLED_CONFIG);
+        if (raw.containsKey(AutoMQConfig.RETRY_STORM_BACKOFF_ENABLED_CONFIG)) {
+            this.enabled = (Boolean) raw.get(AutoMQConfig.RETRY_STORM_BACKOFF_ENABLED_CONFIG);
         }
-        if (configs.containsKey(AutoMQConfig.RETRY_STORM_BACKOFF_MAX_DELAY_MS_CONFIG)) {
-            long nextMaxDelayMs = ConfigUtils.getLong(configs, AutoMQConfig.RETRY_STORM_BACKOFF_MAX_DELAY_MS_CONFIG);
-            validateMaxDelayMs(nextMaxDelayMs);
-            this.maxDelayMs = nextMaxDelayMs;
+        if (raw.containsKey(AutoMQConfig.RETRY_STORM_BACKOFF_MAX_DELAY_MS_CONFIG)) {
+            this.maxDelayMs = (Long) raw.get(AutoMQConfig.RETRY_STORM_BACKOFF_MAX_DELAY_MS_CONFIG);
         }
     }
 

core/src/main/java/kafka/automq/retrystorm/RetryStormBackoffManager.java

@@ -106,7 +106,6 @@ public Set<String> reconfigurableConfigs() {
      */
     @Override
     public void validateReconfiguration(Map<String, ?> configs) throws ConfigException {
-        RetryStormBackoffConfig.validate(configs);
     }
 
     /**

core/src/main/scala/kafka/server/DynamicBrokerConfig.scala

@@ -19,6 +19,7 @@ package kafka.server
 
 import kafka.autobalancer.config.{AutoBalancerControllerConfig, AutoBalancerMetricsReporterConfig}
 import kafka.automq.backpressure.BackPressureConfig
+import kafka.automq.metadata.KafkaGoMetadataResponseRewriter
 import kafka.automq.retrystorm.RetryStormBackoffConfig
 
 import java.util
@@ -105,7 +106,8 @@ object DynamicBrokerConfig {
     AutoBalancerControllerConfig.RECONFIGURABLE_CONFIGS.asScala ++
     AutoBalancerMetricsReporterConfig.RECONFIGURABLE_CONFIGS.asScala ++
     BackPressureConfig.RECONFIGURABLE_CONFIGS.asScala ++
-    RetryStormBackoffConfig.RECONFIGURABLE_CONFIGS.asScala
+    RetryStormBackoffConfig.RECONFIGURABLE_CONFIGS.asScala ++
+    KafkaGoMetadataResponseRewriter.RECONFIGURABLE_CONFIGS.asScala
 
   private val ClusterLevelListenerConfigs = Set(SocketServerConfigs.MAX_CONNECTIONS_CONFIG, SocketServerConfigs.MAX_CONNECTION_CREATION_RATE_CONFIG, SocketServerConfigs.NUM_NETWORK_THREADS_CONFIG)
   private val PerBrokerConfigs = (DynamicSecurityConfigs ++ DynamicListenerConfig.ReconfigurableConfigs).diff(

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -19,6 +19,7 @@ package kafka.server
 
 import kafka.api.ElectLeadersRequestOps
 import kafka.automq.interceptor.ClientIdMetadata
+import kafka.automq.metadata.KafkaGoMetadataResponseRewriter
 import kafka.controller.ReplicaAssignment
 import kafka.coordinator.transaction.{InitProducerIdResult, TransactionCoordinator}
 import kafka.network.RequestChannel
@@ -133,6 +134,8 @@ class KafkaApis(val requestChannel: RequestChannel,
   private val listOffsetSlowExecutor = KafkaApis.newListOffsetExecutor("kafka-apis-list-offset-slow-%d")
   private val listOffsetRequestExecutor = KafkaApis.newListOffsetExecutor("kafka-apis-list-offset-request-%d")
   private val listOffsetInflightPartitions = new AtomicInteger(0)
+  private val kafkaGoMetadataResponseRewriter = new KafkaGoMetadataResponseRewriter(config)
+  config.addReconfigurable(kafkaGoMetadataResponseRewriter)
   // AutoMQ inject end
 
 
@@ -141,6 +144,7 @@ class KafkaApis(val requestChannel: RequestChannel,
     ThreadUtils.shutdownExecutor(listOffsetFastExecutor, 5, TimeUnit.SECONDS, logger.underlying)
     ThreadUtils.shutdownExecutor(listOffsetSlowExecutor, 5, TimeUnit.SECONDS, logger.underlying)
     ThreadUtils.shutdownExecutor(listOffsetRequestExecutor, 5, TimeUnit.SECONDS, logger.underlying)
+    config.removeReconfigurable(kafkaGoMetadataResponseRewriter)
     // AutoMQ inject end
     aclApis.close()
     info("Shutdown complete.")
@@ -1500,6 +1504,10 @@ class KafkaApis(val requestChannel: RequestChannel,
     val completeTopicMetadata =  unknownTopicIdsTopicMetadata ++
       topicMetadata ++ unauthorizedForCreateTopicMetadata ++ unauthorizedForDescribeTopicMetadata
 
+    // AutoMQ inject start
+    kafkaGoMetadataResponseRewriter.rewrite(request.header.clientId, completeTopicMetadata.asJava)
+    // AutoMQ inject end
+
     val brokers = metadataCache.getAliveBrokerNodes(request.context.listenerName)
 
     trace("Sending topic metadata %s and brokers %s for correlation id %d to client %s".format(completeTopicMetadata.mkString(","),

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -802,6 +802,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
   val tableTopicSchemaRegistryUrl = getString(AutoMQConfig.TABLE_TOPIC_SCHEMA_REGISTRY_URL_CONFIG)
   val retryStormBackoffEnabled = getBoolean(AutoMQConfig.RETRY_STORM_BACKOFF_ENABLED_CONFIG)
   val retryStormBackoffMaxDelayMs = getLong(AutoMQConfig.RETRY_STORM_BACKOFF_MAX_DELAY_MS_CONFIG)
+  def kafkaGoMetadataCompatibilityEnabled: Boolean = getBoolean(AutoMQConfig.KAFKA_GO_METADATA_COMPATIBILITY_ENABLED_CONFIG)
   // AutoMQ inject end
 
   /** Internal Configurations **/

core/src/test/java/kafka/automq/backpressure/DefaultBackPressureManagerTest.java

@@ -21,6 +21,9 @@
 
 import kafka.automq.AutoMQConfig;
 
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -29,6 +32,7 @@
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.doAnswer;
@@ -50,6 +54,15 @@ public class DefaultBackPressureManagerTest {
     int schedulerScheduleCalled = 0;
     long schedulerScheduleDelay = 0;
 
+    @Test
+    public void testConfigDefRejectsNegativeCooldown() {
+        ConfigDef configDef = new ConfigDef();
+        AutoMQConfig.define(configDef);
+
+        assertThrows(ConfigException.class, () -> configDef.parse(Map.of(
+            AutoMQConfig.S3_BACK_PRESSURE_COOLDOWN_MS_CONFIG, -1L)));
+    }
+
     @BeforeEach
     public void setup() {
         regulator = mock(Regulator.class);
@@ -89,14 +102,14 @@ public void testDynamicConfig() {
         assertRegulatorCalled(0, 0);
 
         manager.reconfigure(Map.of(
-            AutoMQConfig.S3_BACK_PRESSURE_ENABLED_CONFIG, "true"
+            AutoMQConfig.S3_BACK_PRESSURE_ENABLED_CONFIG, true
         ));
         callChecker(sourceC, LoadLevel.NORMAL);
         callChecker(sourceB, LoadLevel.NORMAL);
         assertRegulatorCalled(1, 1);
 
         manager.reconfigure(Map.of(
-            AutoMQConfig.S3_BACK_PRESSURE_ENABLED_CONFIG, "false"
+            AutoMQConfig.S3_BACK_PRESSURE_ENABLED_CONFIG, false
         ));
         callChecker(sourceC, LoadLevel.NORMAL);
         callChecker(sourceB, LoadLevel.HIGH);