Skip to content

Commit 8d77482

Browse files
superhxsuperhx
committed
feat: add cluster events framework with protobuf-based event publishing
Introduce a cluster-level event system that publishes structured events to the __automq_cluster_events internal topic using CloudEvents + protobuf. Core infrastructure: - ClusterEventPublisher: async, best-effort producer that wraps events as CloudEvents and writes them to the internal topic - ClusterEventsReader: consumer-based reader for retrieving events - ClusterEventsCommand: CLI tool (kafka-cluster-events.sh) for reading and displaying cluster events RequestErrorEvent pipeline: - Detects sustained request error patterns (e.g. auth failures, authorization errors) on the broker and publishes them as events, enabling the control plane to surface risky or misconfigured clients - Integrated into RequestChannel and KafkaApis to capture errors on the broker request path - ResourceErrorExtractor parses per-resource error codes from Kafka API responses (Produce, Fetch, Metadata, etc.) - RequestErrorAccumulator aggregates errors in-memory by (apiKey, errorCode, resource) with client IP/ID tracking, then periodically flushes accumulated buckets as RequestErrorEvent records
1 parent 656e779 commit 8d77482

31 files changed

Lines changed: 2377 additions & 9 deletions

‎.gitignore‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,3 +67,7 @@ __pycache__
6767
bin/
6868
!/bin/
6969
release/.venv/
70+
71+
# Claude Code project config
72+
.claude/settings.local.json
73+
.claude/plans/

‎bin/kafka-cluster-events.sh‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
#!/bin/bash
2+
# Licensed to the Apache Software Foundation (ASF) under one or more
3+
# contributor license agreements. See the NOTICE file distributed with
4+
# this work for additional information regarding copyright ownership.
5+
# The ASF licenses this file to You under the Apache License, Version 2.0
6+
# (the "License"); you may not use this file except in compliance with
7+
# the License. You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing, software
12+
# distributed under the License is distributed on an "AS IS" BASIS,
13+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
# See the License for the specific language governing permissions and
15+
# limitations under the License.
16+
17+
exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.ClusterEventsCommand "$@"

‎build.gradle‎

Lines changed: 32 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ plugins {
4444
// Updating the shadow plugin version to 8.1.1 causes issue with signing and publishing the shadowed
4545
// artifacts - see https://github.com/johnrengelman/shadow/issues/901
4646
id 'com.github.johnrengelman.shadow' version '8.1.0' apply false
47+
id 'com.google.protobuf' version '0.9.4' apply false
4748
// 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
4849
// 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.
4950
// 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
@@ -789,7 +790,7 @@ subprojects {
789790
apply plugin: 'com.diffplug.spotless'
790791
spotless {
791792
java {
792-
targetExclude('src/generated/**/*.java','src/generated-test/**/*.java')
793+
targetExclude('src/generated/**/*.java', 'src/generated-test/**/*.java', 'build/generated/**/*.java')
793794
importOrder('kafka', 'org.apache.kafka', 'com', 'net', 'org', 'java', 'javax', '', '\\#')
794795
removeUnusedImports()
795796
}
@@ -1704,6 +1705,8 @@ project(':generator') {
17041705
}
17051706

17061707
project(':clients') {
1708+
apply plugin: 'com.google.protobuf'
1709+
17071710
base {
17081711
archivesName = "kafka-clients"
17091712
}
@@ -1721,6 +1724,10 @@ project(':clients') {
17211724
implementation libs.opentelemetryProto
17221725
implementation libs.protobuf
17231726

1727+
// AutoMQ inject start
1728+
implementation libs.cloudeventsKafka
1729+
// AutoMQ inject end
1730+
17241731
// libraries which should be added as runtime dependencies in generated pom.xml should be defined here:
17251732
shadowed libs.zstd
17261733
shadowed libs.lz4
@@ -1839,7 +1846,7 @@ project(':clients') {
18391846
sourceSets {
18401847
main {
18411848
java {
1842-
srcDirs = ["src/generated/java", "src/main/java"]
1849+
srcDirs = ["src/generated/java", "src/main/java", "$buildDir/generated/source/proto/main/java"] // AutoMQ: add proto source
18431850
}
18441851
}
18451852
test {
@@ -1849,7 +1856,28 @@ project(':clients') {
18491856
}
18501857
}
18511858

1859+
protobuf {
1860+
protoc {
1861+
artifact = "com.google.protobuf:protoc:$versions.protobuf"
1862+
}
1863+
}
1864+
1865+
// Exclude protobuf-generated sources from checkstyle and spotless
1866+
afterEvaluate {
1867+
checkstyleMain.source = checkstyleMain.source.filter { !it.path.contains('/generated/source/proto/') }
1868+
if (tasks.findByName('spotlessJava')) {
1869+
spotless {
1870+
java {
1871+
targetExclude('build/generated/source/proto/**/*.java')
1872+
}
1873+
}
1874+
}
1875+
}
1876+
18521877
compileJava.dependsOn 'processMessages'
1878+
// AutoMQ inject start: ensure proto sources are generated before compilation
1879+
compileJava.dependsOn 'generateProto'
1880+
// AutoMQ inject end
18531881
srcJar.dependsOn 'processMessages'
18541882

18551883
compileTestJava.dependsOn 'processTestMessages'
@@ -2499,6 +2527,8 @@ project(':tools') {
24992527
implementation (libs.oshi) {
25002528
exclude group: 'org.slf4j', module: 'slf4j-api'
25012529
}
2530+
implementation libs.cloudeventsKafka
2531+
implementation libs.protobuf
25022532
// AutoMQ inject end
25032533

25042534
// for SASL/OAUTHBEARER JWT validation

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1757,6 +1757,14 @@ default ExportClusterManifestResult exportClusterManifest() {
17571757
}
17581758

17591759
ExportClusterManifestResult exportClusterManifest(ExportClusterManifestOptions options);
1760+
1761+
/**
1762+
* Create a reader for the {@code __automq_cluster_events} internal topic.
1763+
*
1764+
* @param sinceMs only return events at or after this epoch-millisecond timestamp, or null for all
1765+
* @return a {@link ClusterEventsReader} that must be closed when done
1766+
*/
1767+
ClusterEventsReader describeClusterEvents(Long sinceMs);
17601768
// AutoMQ inject end
17611769

17621770
/**

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

Lines changed: 173 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,173 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.kafka.clients.admin;
19+
20+
import org.apache.kafka.clients.producer.KafkaProducer;
21+
import org.apache.kafka.clients.producer.ProducerConfig;
22+
import org.apache.kafka.clients.producer.ProducerRecord;
23+
import org.apache.kafka.common.internals.Topic;
24+
import org.apache.kafka.common.serialization.StringSerializer;
25+
26+
import org.slf4j.Logger;
27+
import org.slf4j.LoggerFactory;
28+
29+
import java.net.URI;
30+
import java.time.OffsetDateTime;
31+
import java.util.HashMap;
32+
import java.util.Map;
33+
import java.util.UUID;
34+
import java.util.concurrent.atomic.AtomicBoolean;
35+
import java.util.concurrent.atomic.AtomicReference;
36+
37+
import io.cloudevents.CloudEvent;
38+
import io.cloudevents.core.builder.CloudEventBuilder;
39+
import io.cloudevents.kafka.CloudEventSerializer;
40+
41+
/**
42+
* Publishes cluster events to the {@code __automq_cluster_events} internal topic.
43+
*
44+
* <p>Events are written asynchronously and best-effort: if the producer fails to send an event,
45+
* a warning is logged and the event is dropped. Events must never block or fail the critical path
46+
* (rebalancing, failover, request handling).
47+
*
48+
* <p>Example usage:
49+
* <pre>{@code
50+
* // At broker startup
51+
* ClusterEventPublisher.setup(Map.of("bootstrap.servers", "localhost:9092"));
52+
*
53+
* // From any code path
54+
* ClusterEventPublisher.publish(
55+
* "com.automq.risk.request_error",
56+
* "/automq/broker/0",
57+
* "PRODUCE:my-topic",
58+
* "com.automq.events.RequestErrorEvent",
59+
* requestErrorEvent.toByteArray());
60+
*
61+
* // At shutdown
62+
* ClusterEventPublisher.shutdown();
63+
* }</pre>
64+
*/
65+
public class ClusterEventPublisher implements IClusterEventPublisher {
66+
67+
private static final Logger log = LoggerFactory.getLogger(ClusterEventPublisher.class);
68+
69+
private static final AtomicReference<IClusterEventPublisher> INSTANCE =
70+
new AtomicReference<>(NoopClusterEventPublisher.INSTANCE);
71+
72+
private final KafkaProducer<String, CloudEvent> producer;
73+
private final AtomicBoolean closed = new AtomicBoolean(false);
74+
75+
/**
76+
* Create a publisher from a config map. The map should contain at least
77+
* {@link ProducerConfig#BOOTSTRAP_SERVERS_CONFIG}. Any additional producer or security
78+
* properties (e.g. SASL/SSL) can be included directly. Serializer, acks, retries,
79+
* batch.size and linger.ms are set with sensible defaults if not provided.
80+
*
81+
* @param config producer configuration map
82+
*/
83+
private ClusterEventPublisher(Map<String, Object> config) {
84+
Map<String, Object> props = new HashMap<>(config);
85+
props.putIfAbsent(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
86+
props.putIfAbsent(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CloudEventSerializer.class.getName());
87+
props.putIfAbsent(ProducerConfig.BATCH_SIZE_CONFIG, ClusterEventsConfig.DEFAULT_PUBLISHER_BATCH_SIZE);
88+
props.putIfAbsent(ProducerConfig.LINGER_MS_CONFIG, ClusterEventsConfig.DEFAULT_PUBLISHER_LINGER_MS);
89+
props.putIfAbsent(ProducerConfig.ACKS_CONFIG, "1");
90+
props.putIfAbsent(ProducerConfig.RETRIES_CONFIG, 3);
91+
92+
this.producer = new KafkaProducer<>(props);
93+
}
94+
95+
// ---- Global singleton ----
96+
97+
/**
98+
* Initialize the global singleton publisher. If already set up, the previous instance
99+
* is closed and replaced.
100+
*
101+
* @param config producer configuration map
102+
*/
103+
public static void setup(Map<String, Object> config) {
104+
IClusterEventPublisher prev = INSTANCE.getAndSet(new ClusterEventPublisher(config));
105+
if (prev != null) {
106+
prev.close();
107+
}
108+
}
109+
110+
/**
111+
* Shut down the global singleton publisher, reverting to the no-op instance.
112+
*/
113+
public static void shutdown() {
114+
IClusterEventPublisher prev = INSTANCE.getAndSet(NoopClusterEventPublisher.INSTANCE);
115+
if (prev != null) {
116+
prev.close();
117+
}
118+
}
119+
120+
/**
121+
* Publish an event via the global singleton. If {@link #setup(Map)} has not been called,
122+
* this is a no-op.
123+
*/
124+
public static void publish(String type, String source, String subject, String dataSchema, byte[] data) {
125+
INSTANCE.get().publishEvent(type, source, subject, dataSchema, data);
126+
}
127+
128+
// ---- Instance methods ----
129+
130+
@Override
131+
public void publishEvent(String type, String source, String subject, String dataSchema, byte[] data) {
132+
if (closed.get()) {
133+
log.warn("ClusterEventPublisher is closed, dropping event type={}", type);
134+
return;
135+
}
136+
137+
CloudEvent event = buildEvent(type, source, subject, dataSchema, data);
138+
String key = type + ":" + event.getId();
139+
ProducerRecord<String, CloudEvent> record =
140+
new ProducerRecord<>(Topic.CLUSTER_EVENTS_TOPIC_NAME, key, event);
141+
142+
producer.send(record, (metadata, exception) -> {
143+
if (exception != null) {
144+
log.warn("Failed to publish cluster event type={}, dropping: {}",
145+
record.value().getType(), exception.getMessage());
146+
}
147+
});
148+
}
149+
150+
private CloudEvent buildEvent(String type, String source, String subject, String dataSchema, byte[] data) {
151+
CloudEventBuilder builder = CloudEventBuilder.v1()
152+
.withId(UUID.randomUUID().toString())
153+
.withType(type)
154+
.withSource(URI.create(source))
155+
.withTime(OffsetDateTime.now())
156+
.withDataContentType("application/protobuf")
157+
.withDataSchema(URI.create(dataSchema))
158+
.withData("application/protobuf", data);
159+
160+
if (subject != null) {
161+
builder = builder.withSubject(subject);
162+
}
163+
164+
return builder.build();
165+
}
166+
167+
@Override
168+
public void close() {
169+
if (closed.compareAndSet(false, true)) {
170+
producer.close();
171+
}
172+
}
173+
}

‎clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventTypeRegistry.java‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.kafka.clients.admin;
19+
20+
import org.slf4j.Logger;
21+
import org.slf4j.LoggerFactory;
22+
23+
import java.net.URI;
24+
import java.util.Map;
25+
import java.util.Optional;
26+
import java.util.function.Function;
27+
28+
import io.cloudevents.CloudEvent;
29+
import io.cloudevents.CloudEventData;
30+
31+
/**
32+
* Maps CloudEvent {@code dataschema} values to their POJO wrappers.
33+
*
34+
* <p>Usage:
35+
* <pre>{@code
36+
* ClusterEventTypeRegistry.decode(event).ifPresent(obj -> {
37+
* if (obj instanceof RebalanceSummaryEventData summary) { ... }
38+
* else if (obj instanceof FailoverEventData failover) { ... }
39+
* });
40+
* }</pre>
41+
*/
42+
public class ClusterEventTypeRegistry {
43+
44+
private static final Logger log = LoggerFactory.getLogger(ClusterEventTypeRegistry.class);
45+
46+
private static final Map<String, Function<byte[], Object>> DECODERS = Map.of(
47+
RebalanceSummaryEventData.DATA_SCHEMA, b -> RebalanceSummaryEventData.fromByteArray(b),
48+
RebalancePartitionEventData.DATA_SCHEMA, b -> RebalancePartitionEventData.fromByteArray(b),
49+
FailoverEventData.DATA_SCHEMA, b -> FailoverEventData.fromByteArray(b),
50+
RequestErrorEventData.DATA_SCHEMA, b -> RequestErrorEventData.fromByteArray(b),
51+
OffsetCommitFrequencyEventData.DATA_SCHEMA, b -> OffsetCommitFrequencyEventData.fromByteArray(b)
52+
);
53+
54+
private ClusterEventTypeRegistry() { }
55+
56+
/**
57+
* Decode the protobuf payload of a CloudEvent into its POJO wrapper.
58+
*
59+
* @return the decoded wrapper, or empty if the dataschema is unknown or the payload is invalid
60+
*/
61+
public static Optional<Object> decode(CloudEvent event) {
62+
URI dataSchema = event.getDataSchema();
63+
CloudEventData data = event.getData();
64+
if (dataSchema == null || data == null) {
65+
return Optional.empty();
66+
}
67+
Function<byte[], Object> decoder = DECODERS.get(dataSchema.toString());
68+
if (decoder == null) {
69+
return Optional.empty();
70+
}
71+
try {
72+
return Optional.of(decoder.apply(data.toBytes()));
73+
} catch (Exception e) {
74+
log.debug("Failed to decode CloudEvent payload for dataschema={}", dataSchema, e);
75+
return Optional.empty();
76+
}
77+
}
78+
}

0 commit comments

Comments
 (0)