feat(insepction): add inspection design

.gitignore

@@ -67,3 +67,7 @@ __pycache__
 bin/
 !/bin/
 release/.venv/
+
+# Claude Code project config
+.claude/settings.local.json
+.claude/plans/

bin/kafka-cluster-events.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+# 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.
+
+exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.ClusterEventsCommand "$@"

build.gradle

@@ -44,6 +44,7 @@ plugins {
   // Updating the shadow plugin version to 8.1.1 causes issue with signing and publishing the shadowed
   // artifacts - see https://github.com/johnrengelman/shadow/issues/901
   id 'com.github.johnrengelman.shadow' version '8.1.0' apply false
+  id 'com.google.protobuf' version '0.9.4' apply false
   //  Spotless 6.13.0 has issue with Java 21 (see https://github.com/diffplug/spotless/pull/1920), and Spotless 6.14.0+ requires JRE 11
   //  We are going to drop JDK8 support. Hence, the spotless is upgrade to newest version and be applied only if the build env is compatible with JDK 11.
   //  spotless 6.15.0+ has issue in runtime with JDK8 even through we define it with `apply:false`. see https://github.com/diffplug/spotless/issues/2156 for more details
@@ -157,6 +158,16 @@ allprojects {
     // that are unrelated to the project dependencies, we should not change them
     if (name != "zinc") {
       resolutionStrategy {
+        // AutoMQ inject start
+        eachDependency { details ->
+          if (details.requested.group == 'io.netty' &&
+              details.requested.name != 'netty' &&
+              !details.requested.name.startsWith('netty-tcnative')) {
+            details.useVersion versions.netty
+            details.because('Release packaging must align all Netty modules to a single version')
+          }
+        }
+        // AutoMQ inject end
         force(
           // be explicit about the javassist dependency version instead of relying on the transitive version
           libs.javassist,
@@ -169,8 +180,19 @@ allprojects {
           libs.nettyHandler,
           libs.nettyTransportNativeEpoll,
           // be explicit about the reload4j version instead of relying on the transitive versions
-          libs.reload4j
+          libs.reload4j,
+          // AutoMQ inject start
+          // pin OTel sender versions to prevent version conflicts with Strimzi thirdparty-libs
+          libs.opentelemetryExporterSenderJdk,
+          libs.opentelemetryExporterSenderGrpcManagedChannel
+          // AutoMQ inject end
         )
+        // AutoMQ inject start
+        // swagger is only used for doc generation and still depends on jackson-yaml compatible with SnakeYAML 1.x.
+        if (name != "swagger") {
+          force(libs.snakeyaml)
+        }
+        // AutoMQ inject end
       }
     }
   }
@@ -768,7 +790,7 @@ subprojects {
     apply plugin: 'com.diffplug.spotless'
     spotless {
       java {
-        targetExclude('src/generated/**/*.java','src/generated-test/**/*.java')
+        targetExclude('src/generated/**/*.java', 'src/generated-test/**/*.java', 'build/generated/**/*.java')
         importOrder('kafka', 'org.apache.kafka', 'com', 'net', 'org', 'java', 'javax', '', '\\#')
         removeUnusedImports()
       }
@@ -993,7 +1015,7 @@ project(':core') {
     // table topic start
     implementation ("org.apache.avro:avro:${versions.avro}")
     implementation ("org.apache.avro:avro-protobuf:${versions.avro}")
-    implementation('com.google.protobuf:protobuf-java:3.25.5')
+    implementation libs.protobuf
     implementation ("org.apache.iceberg:iceberg-core:${versions.iceberg}")
     implementation ("org.apache.iceberg:iceberg-api:${versions.iceberg}")
     implementation ("org.apache.iceberg:iceberg-data:${versions.iceberg}")
@@ -1003,7 +1025,7 @@ project(':core') {
     implementation ("org.apache.iceberg:iceberg-nessie:${versions.iceberg}")
     implementation ("software.amazon.awssdk:glue:${versions.awsSdk}")
     implementation ("software.amazon.awssdk:s3tables:${versions.awsSdk}")
-    implementation 'software.amazon.s3tables:s3-tables-catalog-for-iceberg:0.1.0'
+    implementation 'software.amazon.s3tables:s3-tables-catalog-for-iceberg:0.1.8'
 
     implementation ('org.apache.hadoop:hadoop-common:3.4.1') {
       exclude group: 'org.eclipse.jetty', module: '*'
@@ -1051,7 +1073,7 @@ project(':core') {
     //   Wire Runtime for schema handling
     implementation ("com.squareup.wire:wire-schema:${versions.wire}")
     implementation ("com.squareup.wire:wire-runtime:${versions.wire}")
-    implementation 'com.google.api.grpc:proto-google-common-protos:2.52.0'
+    implementation libs.protoGoogleCommonProtos
     // > Protobuf ext end
 
     // table topic end
@@ -1683,6 +1705,8 @@ project(':generator') {
 }
 
 project(':clients') {
+  apply plugin: 'com.google.protobuf'
+
   base {
     archivesName = "kafka-clients"
   }
@@ -1700,6 +1724,10 @@ project(':clients') {
     implementation libs.opentelemetryProto
     implementation libs.protobuf
 
+    // AutoMQ inject start
+    implementation libs.cloudeventsKafka
+    // AutoMQ inject end
+
     // libraries which should be added as runtime dependencies in generated pom.xml should be defined here:
     shadowed libs.zstd
     shadowed libs.lz4
@@ -1818,7 +1846,7 @@ project(':clients') {
   sourceSets {
     main {
       java {
-        srcDirs = ["src/generated/java", "src/main/java"]
+        srcDirs = ["src/generated/java", "src/main/java", "$buildDir/generated/source/proto/main/java"] // AutoMQ: add proto source
       }
     }
     test {
@@ -1828,7 +1856,28 @@ project(':clients') {
     }
   }
 
+  protobuf {
+    protoc {
+      artifact = "com.google.protobuf:protoc:$versions.protobuf"
+    }
+  }
+
+  // Exclude protobuf-generated sources from checkstyle and spotless
+  afterEvaluate {
+    checkstyleMain.source = checkstyleMain.source.filter { !it.path.contains('/generated/source/proto/') }
+    if (tasks.findByName('spotlessJava')) {
+      spotless {
+        java {
+          targetExclude('build/generated/source/proto/**/*.java')
+        }
+      }
+    }
+  }
+
   compileJava.dependsOn 'processMessages'
+  // AutoMQ inject start: ensure proto sources are generated before compilation
+  compileJava.dependsOn 'generateProto'
+  // AutoMQ inject end
   srcJar.dependsOn 'processMessages'
 
   compileTestJava.dependsOn 'processTestMessages'
@@ -2363,6 +2412,7 @@ project(':automq-metrics') {
     api libs.opentelemetryExporterLogging
     api libs.opentelemetryExporterProm
     api libs.opentelemetryExporterOTLP
+    api libs.opentelemetryExporterSenderJdk
     api libs.opentelemetryJmx
 
     // Logging dependencies
@@ -2404,7 +2454,7 @@ project(':automq-metrics') {
     implementation('io.opentelemetry:opentelemetry-sdk:1.40.0')
     implementation("io.opentelemetry.semconv:opentelemetry-semconv:1.25.0-alpha")
     implementation("io.opentelemetry.instrumentation:opentelemetry-runtime-telemetry-java8:2.6.0-alpha")
-    implementation('com.google.protobuf:protobuf-java:3.25.5')
+    implementation libs.protobuf
     implementation('org.xerial.snappy:snappy-java:1.1.10.5')
   }
 
@@ -2474,7 +2524,11 @@ project(':tools') {
       exclude group: 'org.apache.kafka', module: 'kafka-clients'
     }
     implementation libs.bucket4j
-    implementation libs.oshi
+    implementation (libs.oshi) {
+      exclude group: 'org.slf4j', module: 'slf4j-api'
+    }
+    implementation libs.cloudeventsKafka
+    implementation libs.protobuf
     // AutoMQ inject end
 
     // for SASL/OAUTHBEARER JWT validation

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

@@ -1757,6 +1757,14 @@ default ExportClusterManifestResult exportClusterManifest() {
     }
 
     ExportClusterManifestResult exportClusterManifest(ExportClusterManifestOptions options);
+
+    /**
+     * Create a reader for the {@code __automq_cluster_events} internal topic.
+     *
+     * @param sinceMs only return events at or after this epoch-millisecond timestamp, or null for all
+     * @return a {@link ClusterEventsReader} that must be closed when done
+     */
+    ClusterEventsReader describeClusterEvents(Long sinceMs);
     // AutoMQ inject end
 
     /**

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

@@ -0,0 +1,173 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URI;
+import java.time.OffsetDateTime;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import io.cloudevents.kafka.CloudEventSerializer;
+
+/**
+ * Publishes cluster events to the {@code __automq_cluster_events} internal topic.
+ *
+ * <p>Events are written asynchronously and best-effort: if the producer fails to send an event,
+ * a warning is logged and the event is dropped. Events must never block or fail the critical path
+ * (rebalancing, failover, request handling).
+ *
+ * <p>Example usage:
+ * <pre>{@code
+ * // At broker startup
+ * ClusterEventPublisher.setup(Map.of("bootstrap.servers", "localhost:9092"));
+ *
+ * // From any code path
+ * ClusterEventPublisher.publish(
+ *     "com.automq.risk.request_error",
+ *     "/automq/broker/0",
+ *     "PRODUCE:my-topic",
+ *     "com.automq.events.RequestErrorEvent",
+ *     requestErrorEvent.toByteArray());
+ *
+ * // At shutdown
+ * ClusterEventPublisher.shutdown();
+ * }</pre>
+ */
+public class ClusterEventPublisher implements IClusterEventPublisher {
+
+    private static final Logger log = LoggerFactory.getLogger(ClusterEventPublisher.class);
+
+    private static final AtomicReference<IClusterEventPublisher> INSTANCE =
+        new AtomicReference<>(NoopClusterEventPublisher.INSTANCE);
+
+    private final KafkaProducer<String, CloudEvent> producer;
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+
+    /**
+     * Create a publisher from a config map. The map should contain at least
+     * {@link ProducerConfig#BOOTSTRAP_SERVERS_CONFIG}. Any additional producer or security
+     * properties (e.g. SASL/SSL) can be included directly. Serializer, acks, retries,
+     * batch.size and linger.ms are set with sensible defaults if not provided.
+     *
+     * @param config producer configuration map
+     */
+    private ClusterEventPublisher(Map<String, Object> config) {
+        Map<String, Object> props = new HashMap<>(config);
+        props.putIfAbsent(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        props.putIfAbsent(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CloudEventSerializer.class.getName());
+        props.putIfAbsent(ProducerConfig.BATCH_SIZE_CONFIG, ClusterEventsConfig.DEFAULT_PUBLISHER_BATCH_SIZE);
+        props.putIfAbsent(ProducerConfig.LINGER_MS_CONFIG, ClusterEventsConfig.DEFAULT_PUBLISHER_LINGER_MS);
+        props.putIfAbsent(ProducerConfig.ACKS_CONFIG, "1");
+        props.putIfAbsent(ProducerConfig.RETRIES_CONFIG, 3);
+
+        this.producer = new KafkaProducer<>(props);
+    }
+
+    // ---- Global singleton ----
+
+    /**
+     * Initialize the global singleton publisher. If already set up, the previous instance
+     * is closed and replaced.
+     *
+     * @param config producer configuration map
+     */
+    public static void setup(Map<String, Object> config) {
+        IClusterEventPublisher prev = INSTANCE.getAndSet(new ClusterEventPublisher(config));
+        if (prev != null) {
+            prev.close();
+        }
+    }
+
+    /**
+     * Shut down the global singleton publisher, reverting to the no-op instance.
+     */
+    public static void shutdown() {
+        IClusterEventPublisher prev = INSTANCE.getAndSet(NoopClusterEventPublisher.INSTANCE);
+        if (prev != null) {
+            prev.close();
+        }
+    }
+
+    /**
+     * Publish an event via the global singleton. If {@link #setup(Map)} has not been called,
+     * this is a no-op.
+     */
+    public static void publish(String type, String source, String subject, String dataSchema, byte[] data) {
+        INSTANCE.get().publishEvent(type, source, subject, dataSchema, data);
+    }
+
+    // ---- Instance methods ----
+
+    @Override
+    public void publishEvent(String type, String source, String subject, String dataSchema, byte[] data) {
+        if (closed.get()) {
+            log.warn("ClusterEventPublisher is closed, dropping event type={}", type);
+            return;
+        }
+
+        CloudEvent event = buildEvent(type, source, subject, dataSchema, data);
+        String key = type + ":" + event.getId();
+        ProducerRecord<String, CloudEvent> record =
+            new ProducerRecord<>(Topic.CLUSTER_EVENTS_TOPIC_NAME, key, event);
+
+        producer.send(record, (metadata, exception) -> {
+            if (exception != null) {
+                log.warn("Failed to publish cluster event type={}, dropping: {}",
+                    record.value().getType(), exception.getMessage());
+            }
+        });
+    }
+
+    private CloudEvent buildEvent(String type, String source, String subject, String dataSchema, byte[] data) {
+        CloudEventBuilder builder = CloudEventBuilder.v1()
+            .withId(UUID.randomUUID().toString())
+            .withType(type)
+            .withSource(URI.create(source))
+            .withTime(OffsetDateTime.now())
+            .withDataContentType("application/protobuf")
+            .withDataSchema(URI.create(dataSchema))
+            .withData("application/protobuf", data);
+
+        if (subject != null) {
+            builder = builder.withSubject(subject);
+        }
+
+        return builder.build();
+    }
+
+    @Override
+    public void close() {
+        if (closed.compareAndSet(false, true)) {
+            producer.close();
+        }
+    }
+}