test(cluster-events): use protobuf serialization in integration test

Also fix build: include proto-generated sources in clients compilation,
exclude them from spotless/checkstyle/spotbugs.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: superhx

build.gradle

@@ -769,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()
       }
@@ -1826,6 +1826,7 @@ project(':clients') {
     main {
       java {
         srcDirs = ["src/generated/java", "src/main/java"]
+        srcDir "${protobuf.generatedFilesBaseDir}/main/java"
       }
     }
     test {
@@ -1841,6 +1842,18 @@ project(':clients') {
     }
   }
 
+  // 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'
   srcJar.dependsOn 'processMessages'
 
@@ -2489,6 +2502,7 @@ project(':tools') {
     implementation libs.bucket4j
     implementation libs.oshi
     implementation libs.cloudeventsKafka
+    implementation libs.protobuf
     // AutoMQ inject end
 
     // for SASL/OAUTHBEARER JWT validation

gradle/spotbugs-exclude.xml

@@ -636,6 +636,10 @@ For a detailed description of spotbugs bug categories, see https://spotbugs.read
             <Bug pattern="RCN_REDUNDANT_NULLCHECK_OF_NULL_VALUE"/>
         </Or>
     </Match>
+    <!-- Suppress all warnings for protobuf-generated classes -->
+    <Match>
+        <Package name="~com\.automq\.events"/>
+    </Match>
     <!-- AutoMQ inject end -->
 
     <Match>

tools/src/main/java/org/apache/kafka/tools/ClusterEventsIntegrationTest.java

@@ -25,7 +25,9 @@
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.common.errors.TopicExistsException;
 
-import java.nio.charset.StandardCharsets;
+import com.automq.events.RebalanceSummaryEvent;
+import com.automq.events.RequestErrorEvent;
+
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -67,21 +69,30 @@ public static void main(String[] args) throws Exception {
 
         // Step 2: publish test events via ClusterEventPublisher
         try (ClusterEventPublisher publisher = new ClusterEventPublisher(bootstrapServers, Map.of())) {
+            RebalanceSummaryEvent rebalanceEvent = RebalanceSummaryEvent.newBuilder()
+                .setRebalanceId("test-001")
+                .setTriggerReason("load_imbalance")
+                .setPartitionCount(5)
+                .build();
             publisher.publishEvent(
                 "com.automq.ops.rebalance.summary",
                 "/automq/broker/0",
                 "rebalance-test-001",
-                "com.automq.events.RebalanceSummaryEvent",
-                "{\"rebalance_id\":\"test-001\",\"trigger_reason\":\"load_imbalance\",\"partition_count\":5}"
-                    .getBytes(StandardCharsets.UTF_8));
-
+                RebalanceSummaryEvent.getDescriptor().getFullName(),
+                rebalanceEvent.toByteArray());
+
+            RequestErrorEvent requestErrorEvent = RequestErrorEvent.newBuilder()
+                .setApiKey(0)
+                .setErrorCode(29)
+                .setResource("test-topic")
+                .setRps(42.0)
+                .build();
             publisher.publishEvent(
                 "com.automq.risk.request_error",
                 "/automq/broker/0",
                 "PRODUCE:test-topic",
-                "com.automq.events.RequestErrorEvent",
-                "{\"api_key\":0,\"error_code\":29,\"resource\":\"test-topic\",\"rps\":42.0}"
-                    .getBytes(StandardCharsets.UTF_8));
+                RequestErrorEvent.getDescriptor().getFullName(),
+                requestErrorEvent.toByteArray());
 
             System.out.println("✓ Published 2 test events");
         }
@@ -97,6 +108,7 @@ public static void main(String[] args) throws Exception {
             for (CloudEvent e : events) {
                 System.out.printf("  type=%-45s source=%-25s subject=%s%n",
                     e.getType(), e.getSource(), e.getSubject());
+                printEventData(e);
             }
 
             List<CloudEvent> rebalanceEvents = admin.describeClusterEvents(
@@ -118,4 +130,21 @@ public static void main(String[] args) throws Exception {
             System.out.println("\n✓ All assertions passed");
         }
     }
+
+    private static void printEventData(CloudEvent event) throws Exception {
+        io.cloudevents.CloudEventData data = event.getData();
+        if (data == null) return;
+        byte[] bytes = data.toBytes();
+        if (bytes == null) return;
+        String schema = event.getDataSchema() != null ? event.getDataSchema().toString() : "";
+        if (schema.endsWith("RebalanceSummaryEvent")) {
+            RebalanceSummaryEvent msg = RebalanceSummaryEvent.parseFrom(bytes);
+            System.out.printf("    rebalance_id=%s trigger_reason=%s partition_count=%d%n",
+                msg.getRebalanceId(), msg.getTriggerReason(), msg.getPartitionCount());
+        } else if (schema.endsWith("RequestErrorEvent")) {
+            RequestErrorEvent msg = RequestErrorEvent.parseFrom(bytes);
+            System.out.printf("    api_key=%d error_code=%d resource=%s rps=%.1f%n",
+                msg.getApiKey(), msg.getErrorCode(), msg.getResource(), msg.getRps());
+        }
+    }
 }