feat: cluster events

Signed-off-by: Robin Han <hanxvdovehx@gmail.com>

Author: superhx

.claude/CLAUDE.md

@@ -0,0 +1,105 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Build & Test Commands
+
+```bash
+# Full build (skip tests)
+./gradlew build -x test
+
+# Run all tests in a module
+./gradlew :clients:test
+
+# Run a specific test class
+./gradlew :clients:test --tests org.apache.kafka.clients.admin.KafkaAdminClientTest
+
+# Run a specific test method
+./gradlew :clients:test --tests org.apache.kafka.clients.admin.KafkaAdminClientTest.testListTopics
+
+# Checkstyle (linting)
+./gradlew :clients:checkstyleMain :clients:checkstyleTest
+
+# Auto-fix import ordering and formatting
+./gradlew :clients:spotlessApply
+
+# Regenerate Kafka message classes (runs automatically before compileJava)
+./gradlew :clients:processMessages
+```
+
+Java 17 is required. Gradle wrapper is at `./gradlew`.
+
+## Module Structure
+
+This is AutoMQ, a fork of Apache Kafka that replaces the local storage layer with S3-based shared storage. The key modules:
+
+- `clients` — Kafka client libraries (producer, consumer, AdminClient). All new AutoMQ admin APIs go here.
+- `core` — Core broker logic (Scala). Handles request processing, log management, replication.
+- `server` / `server-common` — Broker server implementation and shared utilities.
+- `s3stream` — AutoMQ's S3 storage engine. Replaces Kafka's local log storage.
+- `storage` — Storage abstraction layer; `storage:storage-api` defines the interface.
+- `metadata` / `raft` — KRaft metadata and consensus.
+- `group-coordinator` / `transaction-coordinator` — Coordinator logic extracted from core.
+- `tools` — CLI tools (shell scripts in `bin/` delegate to Java classes here).
+- `automq-shell` — AutoMQ-specific CLI commands.
+
+## AutoMQ Inject Pattern
+
+AutoMQ customizations inside upstream Kafka files are wrapped with marker comments:
+
+```java
+// AutoMQ inject start
+// ... AutoMQ-specific code ...
+// AutoMQ inject end
+```
+
+When adding new AutoMQ functionality to an existing Kafka class, always use these markers. This makes it easy to track divergence from upstream and resolve merge conflicts.
+
+To find all injection points: `grep -rn "AutoMQ inject" --include="*.java"`.
+
+## Code Generation
+
+Several modules generate Java source from JSON message definitions:
+
+- **Message definitions**: `src/main/resources/common/message/*.json`
+- **Generated output**: `src/generated/java/org/apache/kafka/common/message/`
+- **Task**: `processMessages` (runs automatically before `compileJava`)
+
+Do not edit files under `src/generated/` — they are overwritten on each build.
+
+## Adding a New AdminClient Method
+
+The pattern used throughout the codebase:
+
+1. Add `default` + abstract method pair to `Admin.java` inside the `// AutoMQ inject` block.
+2. Add a stub `@Override` to `ForwardingAdmin.java` and `MockAdminClient.java` (test) inside their inject blocks — the build will fail with a compile error if you miss these.
+3. Create `*Options` (extends `AbstractOptions<T>`) and `*Result` (wraps `KafkaFuture`) classes in `clients/src/main/java/org/apache/kafka/clients/admin/`.
+4. Implement in `KafkaAdminClient.java` inside the inject block. For RPC-based methods use the `Call` pattern; for client-side reads (like `describeClusterEvents`) use a dedicated helper class.
+
+## Dependency Management
+
+Dependencies are declared in `gradle/dependencies.gradle` (not `build.gradle`):
+
+```groovy
+// In versions map:
+myLib: "1.0.0"
+
+// In libs map:
+myLib: "com.example:my-lib:$versions.myLib"
+```
+
+Then referenced in `build.gradle` as `implementation libs.myLib`. The `clients` module uses a shadow jar — dependencies added there are shaded; `com.google.protobuf` is relocated to `org.apache.kafka.shaded.com.google.protobuf`, so avoid passing protobuf `Message` objects across the clients module boundary.
+
+## Checkstyle & Spotless
+
+Checkstyle enforces:
+- Max cyclomatic complexity: 16
+- Max NPath complexity: 500
+- No `toLowerCase()`/`toUpperCase()` without a `Locale` argument
+- Import ordering (run `spotlessApply` to fix automatically)
+
+When adding new files, run `./gradlew :module:spotlessApply` before building to avoid import-order failures.
+
+## Integration Testing
+
+If your change or new feature requires integration testing, you can launch a local cluster by following the commands in `devkit/README.md`.

.claude/settings.json

@@ -0,0 +1,22 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(./gradlew *)",
+      "Bash(just *)"
+    ]
+  },
+  "hooks": {
+    "Stop": [
+      {
+        "hooks": [
+          {
+            "type": "command",
+            "command": "cd /Users/robin/code/automq && ./gradlew build -x test 2>&1 | tail -5 | jq -Rs '{systemMessage: (\"Build check: \" + (if test(\"BUILD SUCCESSFUL\") then \"✓ passed\" else \"✗ FAILED — run ./gradlew build -x test for details\" end))}'",
+            "timeout": 300,
+            "statusMessage": "Checking build..."
+          }
+        ]
+      }
+    ]
+  }
+}

.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
@@ -768,7 +769,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()
       }
@@ -1683,6 +1684,8 @@ project(':generator') {
 }
 
 project(':clients') {
+  apply plugin: 'com.google.protobuf'
+
   base {
     archivesName = "kafka-clients"
   }
@@ -1700,6 +1703,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 +1825,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 +1835,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'
@@ -2475,7 +2503,11 @@ project(':tools') {
     }
     implementation libs.bucket4j
     implementation libs.oshi
-    // AutoMQ inject end
+    implementation libs.cloudeventsKafka
+    implementation libs.protobuf
+    // // Proto-generated event classes (separate artifact, excluded from kafka-clients shadow jar)
+    // implementation project(path: ':clients', configuration: 'eventsProto')
+    // // AutoMQ inject end
 
     // for SASL/OAUTHBEARER JWT validation
     implementation (libs.jose4j){
@@ -2538,6 +2570,9 @@ project(':tools') {
     from (configurations.runtimeClasspath) {
       exclude('kafka-clients*')
     }
+    // // AutoMQ inject start
+    // from (project(':clients').configurations.eventsProto.artifacts.files)
+    // // AutoMQ inject end
     into "$buildDir/dependant-libs-${versions.scala}"
     duplicatesStrategy 'exclude'
   }

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

@@ -1757,6 +1757,21 @@ default ExportClusterManifestResult exportClusterManifest() {
     }
 
     ExportClusterManifestResult exportClusterManifest(ExportClusterManifestOptions options);
+
+    /**
+     * Read events from the {@code __cluster_events} internal topic, using the default options.
+     */
+    default DescribeClusterEventsResult describeClusterEvents() {
+        return describeClusterEvents(new DescribeClusterEventsOptions());
+    }
+
+    /**
+     * Read events from the {@code __cluster_events} internal topic.
+     *
+     * @param options Filters and limits for the query.
+     * @return The DescribeClusterEventsResult.
+     */
+    DescribeClusterEventsResult describeClusterEvents(DescribeClusterEventsOptions options);
     // AutoMQ inject end
 
     /**