decoder = DECODERS.get(dataSchema.toString());
+ if (decoder == null) {
+ return Optional.empty();
+ }
+ try {
+ return Optional.of(decoder.apply(data.toBytes()));
+ } catch (Exception e) {
+ log.debug("Failed to decode CloudEvent payload for dataschema={}", dataSchema, e);
+ return Optional.empty();
+ }
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventsConfig.java b/clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventsConfig.java
new file mode 100644
index 0000000000..0edde8027a
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventsConfig.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+/**
+ * Constants for the cluster events system.
+ */
+public class ClusterEventsConfig {
+
+ public static final int DEFAULT_PARTITIONS = 1;
+ public static final long DEFAULT_RETENTION_MS = 7 * 24 * 60 * 60 * 1000L; // 7 days
+ public static final int DEFAULT_PUBLISHER_BATCH_SIZE = 16384;
+ public static final long DEFAULT_PUBLISHER_LINGER_MS = 100L;
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventsReader.java b/clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventsReader.java
new file mode 100644
index 0000000000..d0c16fdde8
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventsReader.java
@@ -0,0 +1,133 @@
+/*
+ * 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.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+import java.io.Closeable;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.kafka.CloudEventDeserializer;
+
+/**
+ * Reads events from the {@code __automq_cluster_events} internal topic.
+ *
+ * Supports both one-shot reads and continuous polling (tail mode).
+ * Call {@link #poll(Duration)} repeatedly to consume new events as they arrive.
+ * Must be {@link #close() closed} when done.
+ */
+public class ClusterEventsReader implements Closeable {
+
+ private static final TopicPartition TP =
+ new TopicPartition(Topic.CLUSTER_EVENTS_TOPIC_NAME, 0);
+
+ private final KafkaConsumer consumer;
+ private long endOffset;
+
+ ClusterEventsReader(Map consumerConfig, Long sinceMs) {
+ this.consumer = createConsumer(consumerConfig);
+ consumer.assign(Collections.singletonList(TP));
+ seek(sinceMs);
+ refreshEndOffset();
+ }
+
+ /**
+ * Poll for new events.
+ *
+ * @return events received within the timeout, may be empty
+ */
+ public List poll(Duration timeout) {
+ List results = new ArrayList<>();
+ for (ConsumerRecord record : consumer.poll(timeout)) {
+ if (record.value() != null) {
+ results.add(record.value());
+ }
+ }
+ return results;
+ }
+
+ /**
+ * Returns true if the consumer position has reached the end offset
+ * that was snapshotted when this reader was created.
+ */
+ public boolean isPolledToEnd() {
+ return consumer.position(TP) >= endOffset;
+ }
+
+ @Override
+ public void close() {
+ consumer.close();
+ }
+
+ private void refreshEndOffset() {
+ Long end = consumer.endOffsets(Collections.singletonList(TP)).get(TP);
+ this.endOffset = end != null ? end : 0L;
+ }
+
+ private void seek(Long sinceMs) {
+ if (sinceMs != null) {
+ Map offsets =
+ consumer.offsetsForTimes(Collections.singletonMap(TP, sinceMs));
+ OffsetAndTimestamp ot = offsets.get(TP);
+ if (ot != null) {
+ consumer.seek(TP, ot.offset());
+ } else {
+ consumer.seekToEnd(Collections.singletonList(TP));
+ }
+ } else {
+ consumer.seekToBeginning(Collections.singletonList(TP));
+ }
+ }
+
+ private static KafkaConsumer createConsumer(Map config) {
+ Properties props = new Properties();
+ props.putAll(config);
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, CloudEventDeserializer.class.getName());
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+ return new KafkaConsumer<>(props);
+ }
+
+ static Map buildConsumerConfig(AdminClientConfig config) {
+ Map result = new java.util.HashMap<>();
+ result.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
+ String.join(",", config.getList(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG)));
+ for (Map.Entry entry : config.values().entrySet()) {
+ String key = entry.getKey();
+ Object value = entry.getValue();
+ if (value != null && (key.startsWith("ssl.") || key.startsWith("sasl.")
+ || key.equals("security.protocol"))) {
+ result.put(key, value);
+ }
+ }
+ return result;
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/FailoverEventData.java b/clients/src/main/java/org/apache/kafka/clients/admin/FailoverEventData.java
new file mode 100644
index 0000000000..e4435d1442
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/FailoverEventData.java
@@ -0,0 +1,102 @@
+/*
+ * 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 com.automq.events.FailoverEvent;
+
+/**
+ * POJO wrapper for {@link FailoverEvent} protobuf message.
+ */
+public class FailoverEventData {
+
+ public static final String TYPE = "com.automq.ops.failover";
+ public static final String DATA_SCHEMA = FailoverEvent.getDescriptor().getFullName();
+
+ private int failedNodeId;
+ private long detectedTimestamp;
+ private long completedTimestamp;
+ private String detail;
+
+ public FailoverEventData setFailedNodeId(int failedNodeId) {
+ this.failedNodeId = failedNodeId;
+ return this;
+ }
+
+ public FailoverEventData setDetectedTimestamp(long detectedTimestamp) {
+ this.detectedTimestamp = detectedTimestamp;
+ return this;
+ }
+
+ public FailoverEventData setCompletedTimestamp(long completedTimestamp) {
+ this.completedTimestamp = completedTimestamp;
+ return this;
+ }
+
+ public FailoverEventData setDetail(String detail) {
+ this.detail = detail;
+ return this;
+ }
+
+ public int failedNodeId() {
+ return failedNodeId;
+ }
+
+ public long detectedTimestamp() {
+ return detectedTimestamp;
+ }
+
+ public long completedTimestamp() {
+ return completedTimestamp;
+ }
+
+ public String detail() {
+ return detail;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("FailoverEvent{");
+ sb.append("failedNodeId=").append(failedNodeId);
+ sb.append(", detectedTimestamp=").append(detectedTimestamp);
+ sb.append(", completedTimestamp=").append(completedTimestamp);
+ if (detail != null) sb.append(", detail='").append(detail).append('\'');
+ sb.append('}');
+ return sb.toString();
+ }
+
+ public byte[] toByteArray() {
+ FailoverEvent.Builder b = FailoverEvent.newBuilder()
+ .setFailedNodeId(failedNodeId)
+ .setDetectedTimestamp(detectedTimestamp)
+ .setCompletedTimestamp(completedTimestamp);
+ if (detail != null) b.setDetail(detail);
+ return b.build().toByteArray();
+ }
+
+ public static FailoverEventData fromByteArray(byte[] data) {
+ try {
+ FailoverEvent e = FailoverEvent.parseFrom(data);
+ return new FailoverEventData()
+ .setFailedNodeId(e.getFailedNodeId())
+ .setDetectedTimestamp(e.getDetectedTimestamp())
+ .setCompletedTimestamp(e.getCompletedTimestamp())
+ .setDetail(e.getDetail());
+ } catch (Exception ex) {
+ throw new IllegalArgumentException("Failed to parse FailoverEvent", ex);
+ }
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ForwardingAdmin.java b/clients/src/main/java/org/apache/kafka/clients/admin/ForwardingAdmin.java
index 5b61e99ea2..e3d1bba931 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/ForwardingAdmin.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ForwardingAdmin.java
@@ -335,5 +335,10 @@ public ExportClusterManifestResult exportClusterManifest(ExportClusterManifestOp
return delegate.exportClusterManifest(options);
}
+ @Override
+ public ClusterEventsReader describeClusterEvents(Long sinceMs) {
+ return delegate.describeClusterEvents(sinceMs);
+ }
+
// AutoMQ inject end
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/IClusterEventPublisher.java b/clients/src/main/java/org/apache/kafka/clients/admin/IClusterEventPublisher.java
new file mode 100644
index 0000000000..116bc610da
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/IClusterEventPublisher.java
@@ -0,0 +1,38 @@
+/*
+ * 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;
+
+/**
+ * Interface for publishing cluster events to the {@code __automq_cluster_events} internal topic.
+ */
+public interface IClusterEventPublisher extends AutoCloseable {
+
+ /**
+ * Publish a domain event to {@code __automq_cluster_events}.
+ *
+ * @param type CloudEvent type (reverse-DNS), e.g. {@code "com.automq.ops.rebalance.summary"}
+ * @param source CloudEvent source URI, e.g. {@code "/automq/broker/0"}
+ * @param subject CloudEvent subject, or {@code null} if not applicable
+ * @param dataSchema fully-qualified protobuf message type name
+ * @param data serialized protobuf payload
+ */
+ void publishEvent(String type, String source, String subject, String dataSchema, byte[] data);
+
+ @Override
+ void close();
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index addb4ba66c..2bfc4dfaf3 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -420,6 +420,10 @@ public class KafkaAdminClient extends AdminClient {
private final boolean clientTelemetryEnabled;
private final MetadataRecoveryStrategy metadataRecoveryStrategy;
+ // AutoMQ inject start
+ private final AdminClientConfig adminClientConfig;
+ // AutoMQ inject end
+
/**
* The telemetry requests client instance id.
*/
@@ -633,6 +637,9 @@ private KafkaAdminClient(AdminClientConfig config,
CommonClientConfigs.RETRY_BACKOFF_JITTER);
this.clientTelemetryEnabled = config.getBoolean(AdminClientConfig.ENABLE_METRICS_PUSH_CONFIG);
this.metadataRecoveryStrategy = MetadataRecoveryStrategy.forName(config.getString(AdminClientConfig.METADATA_RECOVERY_STRATEGY_CONFIG));
+ // AutoMQ inject start
+ this.adminClientConfig = config;
+ // AutoMQ inject end
config.logUnused();
AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics, time.milliseconds());
log.debug("Kafka admin client initialized");
@@ -2458,7 +2465,7 @@ private TopicDescription getTopicDescriptionFromDescribeTopicsResponseTopic(
return new TopicDescription(topic.name(), topic.isInternal(), partitions, authorisedOperations, topic.topicId());
}
- // AutoMQ for Kafka inject start
+ // AutoMQ inject start
@Override
public GetNextNodeIdResult getNextNodeId(GetNextNodeIdOptions options) {
KafkaFutureImpl nodeIdFuture = new KafkaFutureImpl<>();
@@ -2615,7 +2622,12 @@ void handleFailure(Throwable throwable) {
runnable.call(call, now);
return new ExportClusterManifestResult(future);
}
- // AutoMQ for Kafka inject end
+
+ @Override
+ public ClusterEventsReader describeClusterEvents(Long sinceMs) {
+ return new ClusterEventsReader(ClusterEventsReader.buildConsumerConfig(adminClientConfig), sinceMs);
+ }
+ // AutoMQ inject end
private TopicDescription getTopicDescriptionFromCluster(Cluster cluster, String topicName, Uuid topicId,
Integer authorizedOperations) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/NoopClusterEventPublisher.java b/clients/src/main/java/org/apache/kafka/clients/admin/NoopClusterEventPublisher.java
new file mode 100644
index 0000000000..bc71bbe4ab
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/NoopClusterEventPublisher.java
@@ -0,0 +1,37 @@
+/*
+ * 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;
+
+/**
+ * No-op implementation used when the publisher has not been set up.
+ * All events are silently dropped.
+ */
+public class NoopClusterEventPublisher implements IClusterEventPublisher {
+
+ static final NoopClusterEventPublisher INSTANCE = new NoopClusterEventPublisher();
+
+ @Override
+ public void publishEvent(String type, String source, String subject, String dataSchema, byte[] data) {
+ // no-op
+ }
+
+ @Override
+ public void close() {
+ // no-op
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/OffsetCommitFrequencyEventData.java b/clients/src/main/java/org/apache/kafka/clients/admin/OffsetCommitFrequencyEventData.java
new file mode 100644
index 0000000000..8d4dbd2625
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/OffsetCommitFrequencyEventData.java
@@ -0,0 +1,119 @@
+/*
+ * 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 com.automq.events.OffsetCommitFrequencyEvent;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * POJO wrapper for {@link OffsetCommitFrequencyEvent} protobuf message.
+ */
+public class OffsetCommitFrequencyEventData {
+
+ public static final String TYPE = "com.automq.risk.offset_commit_frequency";
+ public static final String DATA_SCHEMA = OffsetCommitFrequencyEvent.getDescriptor().getFullName();
+
+ private List clientIps = Collections.emptyList();
+ private List clientIds = Collections.emptyList();
+ private String groupId;
+ private String topic;
+ private double rps;
+
+ public OffsetCommitFrequencyEventData setClientIps(List clientIps) {
+ this.clientIps = clientIps != null ? clientIps : Collections.emptyList();
+ return this;
+ }
+
+ public OffsetCommitFrequencyEventData setClientIds(List clientIds) {
+ this.clientIds = clientIds != null ? clientIds : Collections.emptyList();
+ return this;
+ }
+
+ public OffsetCommitFrequencyEventData setGroupId(String groupId) {
+ this.groupId = groupId;
+ return this;
+ }
+
+ public OffsetCommitFrequencyEventData setTopic(String topic) {
+ this.topic = topic;
+ return this;
+ }
+
+ public OffsetCommitFrequencyEventData setRps(double rps) {
+ this.rps = rps;
+ return this;
+ }
+
+ public List clientIps() {
+ return clientIps;
+ }
+
+ public List clientIds() {
+ return clientIds;
+ }
+
+ public String groupId() {
+ return groupId;
+ }
+
+ public String topic() {
+ return topic;
+ }
+
+ public double rps() {
+ return rps;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("OffsetCommitFrequencyEvent{");
+ if (!clientIps.isEmpty()) sb.append("clientIps=").append(clientIps).append(", ");
+ if (!clientIds.isEmpty()) sb.append("clientIds=").append(clientIds).append(", ");
+ sb.append("groupId='").append(groupId).append('\'');
+ sb.append(", topic='").append(topic).append('\'');
+ sb.append(", rps=").append(rps);
+ sb.append('}');
+ return sb.toString();
+ }
+
+ public byte[] toByteArray() {
+ OffsetCommitFrequencyEvent.Builder b = OffsetCommitFrequencyEvent.newBuilder()
+ .addAllClientIps(clientIps)
+ .addAllClientIds(clientIds)
+ .setRps(rps);
+ if (groupId != null) b.setGroupId(groupId);
+ if (topic != null) b.setTopic(topic);
+ return b.build().toByteArray();
+ }
+
+ public static OffsetCommitFrequencyEventData fromByteArray(byte[] data) {
+ try {
+ OffsetCommitFrequencyEvent e = OffsetCommitFrequencyEvent.parseFrom(data);
+ return new OffsetCommitFrequencyEventData()
+ .setClientIps(new ArrayList<>(e.getClientIpsList()))
+ .setClientIds(new ArrayList<>(e.getClientIdsList()))
+ .setGroupId(e.getGroupId())
+ .setTopic(e.getTopic())
+ .setRps(e.getRps());
+ } catch (Exception ex) {
+ throw new IllegalArgumentException("Failed to parse OffsetCommitFrequencyEvent", ex);
+ }
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/RebalancePartitionEventData.java b/clients/src/main/java/org/apache/kafka/clients/admin/RebalancePartitionEventData.java
new file mode 100644
index 0000000000..c38c444e5e
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/RebalancePartitionEventData.java
@@ -0,0 +1,115 @@
+/*
+ * 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 com.automq.events.RebalancePartitionEvent;
+
+/**
+ * POJO wrapper for {@link RebalancePartitionEvent} protobuf message.
+ */
+public class RebalancePartitionEventData {
+
+ public static final String TYPE = "com.automq.ops.rebalance.partition";
+ public static final String DATA_SCHEMA = RebalancePartitionEvent.getDescriptor().getFullName();
+
+ private String rebalanceId;
+ private String topicPartition;
+ private int fromBroker;
+ private int toBroker;
+ private String reason;
+
+ public RebalancePartitionEventData setRebalanceId(String rebalanceId) {
+ this.rebalanceId = rebalanceId;
+ return this;
+ }
+
+ public RebalancePartitionEventData setTopicPartition(String topicPartition) {
+ this.topicPartition = topicPartition;
+ return this;
+ }
+
+ public RebalancePartitionEventData setFromBroker(int fromBroker) {
+ this.fromBroker = fromBroker;
+ return this;
+ }
+
+ public RebalancePartitionEventData setToBroker(int toBroker) {
+ this.toBroker = toBroker;
+ return this;
+ }
+
+ public RebalancePartitionEventData setReason(String reason) {
+ this.reason = reason;
+ return this;
+ }
+
+ public String rebalanceId() {
+ return rebalanceId;
+ }
+
+ public String topicPartition() {
+ return topicPartition;
+ }
+
+ public int fromBroker() {
+ return fromBroker;
+ }
+
+ public int toBroker() {
+ return toBroker;
+ }
+
+ public String reason() {
+ return reason;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("RebalancePartitionEvent{");
+ sb.append("rebalanceId='").append(rebalanceId).append('\'');
+ sb.append(", topicPartition='").append(topicPartition).append('\'');
+ sb.append(", fromBroker=").append(fromBroker);
+ sb.append(", toBroker=").append(toBroker);
+ if (reason != null) sb.append(", reason='").append(reason).append('\'');
+ sb.append('}');
+ return sb.toString();
+ }
+
+ public byte[] toByteArray() {
+ RebalancePartitionEvent.Builder b = RebalancePartitionEvent.newBuilder();
+ if (rebalanceId != null) b.setRebalanceId(rebalanceId);
+ if (topicPartition != null) b.setTopicPartition(topicPartition);
+ b.setFromBroker(fromBroker);
+ b.setToBroker(toBroker);
+ if (reason != null) b.setReason(reason);
+ return b.build().toByteArray();
+ }
+
+ public static RebalancePartitionEventData fromByteArray(byte[] data) {
+ try {
+ RebalancePartitionEvent e = RebalancePartitionEvent.parseFrom(data);
+ return new RebalancePartitionEventData()
+ .setRebalanceId(e.getRebalanceId())
+ .setTopicPartition(e.getTopicPartition())
+ .setFromBroker(e.getFromBroker())
+ .setToBroker(e.getToBroker())
+ .setReason(e.getReason());
+ } catch (Exception ex) {
+ throw new IllegalArgumentException("Failed to parse RebalancePartitionEvent", ex);
+ }
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/RebalanceSummaryEventData.java b/clients/src/main/java/org/apache/kafka/clients/admin/RebalanceSummaryEventData.java
new file mode 100644
index 0000000000..91569232c7
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/RebalanceSummaryEventData.java
@@ -0,0 +1,118 @@
+/*
+ * 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 com.automq.events.RebalanceSummaryEvent;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * POJO wrapper for {@link RebalanceSummaryEvent} protobuf message.
+ */
+public class RebalanceSummaryEventData {
+
+ public static final String TYPE = "com.automq.ops.rebalance.summary";
+ public static final String DATA_SCHEMA = RebalanceSummaryEvent.getDescriptor().getFullName();
+
+ private String rebalanceId;
+ private String triggerReason;
+ private int partitionCount;
+ private Map brokerLoadBefore = new HashMap<>();
+ private Map brokerLoadAfter = new HashMap<>();
+
+ public RebalanceSummaryEventData setRebalanceId(String rebalanceId) {
+ this.rebalanceId = rebalanceId;
+ return this;
+ }
+
+ public RebalanceSummaryEventData setTriggerReason(String triggerReason) {
+ this.triggerReason = triggerReason;
+ return this;
+ }
+
+ public RebalanceSummaryEventData setPartitionCount(int partitionCount) {
+ this.partitionCount = partitionCount;
+ return this;
+ }
+
+ public RebalanceSummaryEventData setBrokerLoadBefore(Map brokerLoadBefore) {
+ this.brokerLoadBefore = brokerLoadBefore != null ? brokerLoadBefore : new HashMap<>();
+ return this;
+ }
+
+ public RebalanceSummaryEventData setBrokerLoadAfter(Map brokerLoadAfter) {
+ this.brokerLoadAfter = brokerLoadAfter != null ? brokerLoadAfter : new HashMap<>();
+ return this;
+ }
+
+ public String rebalanceId() {
+ return rebalanceId;
+ }
+
+ public String triggerReason() {
+ return triggerReason;
+ }
+
+ public int partitionCount() {
+ return partitionCount;
+ }
+
+ public Map brokerLoadBefore() {
+ return brokerLoadBefore;
+ }
+
+ public Map brokerLoadAfter() {
+ return brokerLoadAfter;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("RebalanceSummaryEvent{");
+ sb.append("rebalanceId='").append(rebalanceId).append('\'');
+ sb.append(", triggerReason='").append(triggerReason).append('\'');
+ sb.append(", partitionCount=").append(partitionCount);
+ if (!brokerLoadBefore.isEmpty()) sb.append(", brokerLoadBefore=").append(brokerLoadBefore);
+ if (!brokerLoadAfter.isEmpty()) sb.append(", brokerLoadAfter=").append(brokerLoadAfter);
+ sb.append('}');
+ return sb.toString();
+ }
+
+ public byte[] toByteArray() {
+ RebalanceSummaryEvent.Builder b = RebalanceSummaryEvent.newBuilder();
+ if (rebalanceId != null) b.setRebalanceId(rebalanceId);
+ if (triggerReason != null) b.setTriggerReason(triggerReason);
+ b.setPartitionCount(partitionCount);
+ b.putAllBrokerLoadBefore(brokerLoadBefore);
+ b.putAllBrokerLoadAfter(brokerLoadAfter);
+ return b.build().toByteArray();
+ }
+
+ public static RebalanceSummaryEventData fromByteArray(byte[] data) {
+ try {
+ RebalanceSummaryEvent e = RebalanceSummaryEvent.parseFrom(data);
+ return new RebalanceSummaryEventData()
+ .setRebalanceId(e.getRebalanceId())
+ .setTriggerReason(e.getTriggerReason())
+ .setPartitionCount(e.getPartitionCount())
+ .setBrokerLoadBefore(new HashMap<>(e.getBrokerLoadBeforeMap()))
+ .setBrokerLoadAfter(new HashMap<>(e.getBrokerLoadAfterMap()));
+ } catch (Exception ex) {
+ throw new IllegalArgumentException("Failed to parse RebalanceSummaryEvent", ex);
+ }
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/RequestErrorEventData.java b/clients/src/main/java/org/apache/kafka/clients/admin/RequestErrorEventData.java
new file mode 100644
index 0000000000..63ba309517
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/RequestErrorEventData.java
@@ -0,0 +1,159 @@
+/*
+ * 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 com.automq.events.RequestErrorEvent;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * POJO wrapper for {@link RequestErrorEvent} protobuf message.
+ *
+ * This class hides all protobuf references so that modules with conflicting
+ * protobuf versions (shaded vs unshaded) can build and serialize
+ * {@code RequestErrorEvent} without directly touching protobuf classes.
+ */
+public class RequestErrorEventData {
+
+ public static final String TYPE = "com.automq.risk.request_error";
+ public static final String DATA_SCHEMA = RequestErrorEvent.getDescriptor().getFullName();
+
+ private int apiKey;
+ private int errorCode;
+ private String resource;
+ private List clientIps = Collections.emptyList();
+ private List clientIds = Collections.emptyList();
+ private double rps;
+ private String reason;
+
+ public RequestErrorEventData setApiKey(int apiKey) {
+ this.apiKey = apiKey;
+ return this;
+ }
+
+ public RequestErrorEventData setErrorCode(int errorCode) {
+ this.errorCode = errorCode;
+ return this;
+ }
+
+ public RequestErrorEventData setResource(String resource) {
+ this.resource = resource;
+ return this;
+ }
+
+ public RequestErrorEventData setClientIps(List clientIps) {
+ this.clientIps = clientIps != null ? clientIps : Collections.emptyList();
+ return this;
+ }
+
+ public RequestErrorEventData setClientIds(List clientIds) {
+ this.clientIds = clientIds != null ? clientIds : Collections.emptyList();
+ return this;
+ }
+
+ public RequestErrorEventData setRps(double rps) {
+ this.rps = rps;
+ return this;
+ }
+
+ public RequestErrorEventData setReason(String reason) {
+ this.reason = reason;
+ return this;
+ }
+
+ public int apiKey() {
+ return apiKey;
+ }
+
+ public int errorCode() {
+ return errorCode;
+ }
+
+ public String resource() {
+ return resource;
+ }
+
+ public List clientIps() {
+ return clientIps;
+ }
+
+ public List clientIds() {
+ return clientIds;
+ }
+
+ public double rps() {
+ return rps;
+ }
+
+ public String reason() {
+ return reason;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("RequestErrorEvent{");
+ sb.append("apiKey=").append(apiKey);
+ sb.append(", errorCode=").append(errorCode);
+ if (resource != null) sb.append(", resource='").append(resource).append('\'');
+ if (!clientIps.isEmpty()) sb.append(", clientIps=").append(clientIps);
+ if (!clientIds.isEmpty()) sb.append(", clientIds=").append(clientIds);
+ sb.append(", rps=").append(rps);
+ if (reason != null) sb.append(", reason='").append(reason).append('\'');
+ sb.append('}');
+ return sb.toString();
+ }
+
+ /**
+ * Serialize to protobuf byte array.
+ */
+ public byte[] toByteArray() {
+ RequestErrorEvent.Builder b = RequestErrorEvent.newBuilder()
+ .setApiKey(apiKey)
+ .setErrorCode(errorCode)
+ .addAllClientIps(clientIps)
+ .addAllClientIds(clientIds)
+ .setRps(rps);
+ if (resource != null) {
+ b.setResource(resource);
+ }
+ if (reason != null) {
+ b.setReason(reason);
+ }
+ return b.build().toByteArray();
+ }
+
+ /**
+ * Deserialize from protobuf byte array.
+ */
+ public static RequestErrorEventData fromByteArray(byte[] data) {
+ try {
+ RequestErrorEvent event = RequestErrorEvent.parseFrom(data);
+ return new RequestErrorEventData()
+ .setApiKey(event.getApiKey())
+ .setErrorCode(event.getErrorCode())
+ .setResource(event.hasResource() ? event.getResource() : null)
+ .setClientIps(new ArrayList<>(event.getClientIpsList()))
+ .setClientIds(new ArrayList<>(event.getClientIdsList()))
+ .setRps(event.getRps())
+ .setReason(event.hasReason() ? event.getReason() : null);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Failed to parse RequestErrorEvent", e);
+ }
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/common/internals/Topic.java b/clients/src/main/java/org/apache/kafka/common/internals/Topic.java
index 1069a691b4..88c6e9cfd7 100644
--- a/clients/src/main/java/org/apache/kafka/common/internals/Topic.java
+++ b/clients/src/main/java/org/apache/kafka/common/internals/Topic.java
@@ -35,6 +35,7 @@ public class Topic {
public static final String AUTO_BALANCER_METRICS_TOPIC_NAME = "__auto_balancer_metrics";
public static final String TABLE_TOPIC_CONTROL_TOPIC_NAME = "__automq_table_control";
public static final String TABLE_TOPIC_DATA_TOPIC_NAME = "__automq_table_data";
+ public static final String CLUSTER_EVENTS_TOPIC_NAME = "__automq_cluster_events";
// Full name: __automq_consumer_group_force_commit
public static final String FORCE_COMMIT_SENTINEL_TOPIC = "__a.c.g.f.c";
// AutoMQ inject end
diff --git a/clients/src/main/proto/com/automq/events/cluster_events.proto b/clients/src/main/proto/com/automq/events/cluster_events.proto
new file mode 100644
index 0000000000..00333beee7
--- /dev/null
+++ b/clients/src/main/proto/com/automq/events/cluster_events.proto
@@ -0,0 +1,70 @@
+// 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.
+
+syntax = "proto3";
+
+package com.automq.events;
+
+option java_package = "com.automq.events";
+option java_multiple_files = true;
+option java_outer_classname = "ClusterEventsProto";
+
+// Emitted once per AutoBalancer rebalance action — summary of what moved and why.
+message RebalanceSummaryEvent {
+ string rebalance_id = 1;
+ string trigger_reason = 2;
+ int32 partition_count = 3;
+ map broker_load_before = 4;
+ map broker_load_after = 5;
+}
+
+// Emitted once per partition moved during a rebalance. Correlates with
+// RebalanceSummaryEvent via rebalance_id.
+message RebalancePartitionEvent {
+ string rebalance_id = 1;
+ string topic_partition = 2;
+ int32 from_broker = 3;
+ int32 to_broker = 4;
+ string reason = 5;
+}
+
+// Emitted when a broker failover completes.
+message FailoverEvent {
+ int32 failed_node_id = 1;
+ int64 detected_timestamp = 2;
+ int64 completed_timestamp = 3;
+ string detail = 4;
+}
+
+// Emitted when a broker detects a sustained rate of request errors
+// (e.g. auth failures, authorization errors) for a given API/resource.
+message RequestErrorEvent {
+ int32 api_key = 1;
+ int32 error_code = 2;
+ optional string resource = 3;
+ repeated string client_ips = 4;
+ repeated string client_ids = 5;
+ double rps = 6;
+ optional string reason = 7;
+}
+
+// Emitted when a consumer group is committing offsets at an excessive rate.
+message OffsetCommitFrequencyEvent {
+ repeated string client_ips = 1;
+ repeated string client_ids = 2;
+ string group_id = 3;
+ string topic = 4;
+ double rps = 5;
+}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/ClusterEventTypeRegistryTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/ClusterEventTypeRegistryTest.java
new file mode 100644
index 0000000000..d7dd07d20c
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/ClusterEventTypeRegistryTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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 com.automq.events.FailoverEvent;
+import com.automq.events.OffsetCommitFrequencyEvent;
+import com.automq.events.RebalancePartitionEvent;
+import com.automq.events.RebalanceSummaryEvent;
+import com.automq.events.RequestErrorEvent;
+
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.Optional;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ClusterEventTypeRegistryTest {
+
+ @Test
+ void decodeRebalanceSummaryEvent() {
+ RebalanceSummaryEvent payload = RebalanceSummaryEvent.newBuilder()
+ .setRebalanceId("r1")
+ .setTriggerReason("load imbalance")
+ .setPartitionCount(3)
+ .build();
+
+ CloudEvent event = buildEvent(RebalanceSummaryEventData.DATA_SCHEMA, payload.toByteArray());
+ Optional result = ClusterEventTypeRegistry.decode(event);
+
+ assertTrue(result.isPresent());
+ RebalanceSummaryEventData decoded = assertInstanceOf(RebalanceSummaryEventData.class, result.get());
+ assertEquals("r1", decoded.rebalanceId());
+ assertEquals("load imbalance", decoded.triggerReason());
+ assertEquals(3, decoded.partitionCount());
+ }
+
+ @Test
+ void decodeRebalancePartitionEvent() {
+ RebalancePartitionEvent payload = RebalancePartitionEvent.newBuilder()
+ .setRebalanceId("r1")
+ .setTopicPartition("my-topic-0")
+ .setFromBroker(1)
+ .setToBroker(2)
+ .build();
+
+ CloudEvent event = buildEvent(RebalancePartitionEventData.DATA_SCHEMA, payload.toByteArray());
+ Optional result = ClusterEventTypeRegistry.decode(event);
+
+ assertTrue(result.isPresent());
+ RebalancePartitionEventData decoded = assertInstanceOf(RebalancePartitionEventData.class, result.get());
+ assertEquals("my-topic-0", decoded.topicPartition());
+ assertEquals(1, decoded.fromBroker());
+ assertEquals(2, decoded.toBroker());
+ }
+
+ @Test
+ void decodeFailoverEvent() {
+ FailoverEvent payload = FailoverEvent.newBuilder()
+ .setFailedNodeId(5)
+ .setDetectedTimestamp(1000L)
+ .setCompletedTimestamp(2000L)
+ .setDetail("node unreachable")
+ .build();
+
+ CloudEvent event = buildEvent(FailoverEventData.DATA_SCHEMA, payload.toByteArray());
+ Optional result = ClusterEventTypeRegistry.decode(event);
+
+ assertTrue(result.isPresent());
+ FailoverEventData decoded = assertInstanceOf(FailoverEventData.class, result.get());
+ assertEquals(5, decoded.failedNodeId());
+ assertEquals("node unreachable", decoded.detail());
+ }
+
+ @Test
+ void decodeRequestErrorEvent() {
+ RequestErrorEvent payload = RequestErrorEvent.newBuilder()
+ .setApiKey(0)
+ .setErrorCode(29)
+ .setResource("my-topic")
+ .setRps(42.0)
+ .build();
+
+ CloudEvent event = buildEvent(RequestErrorEventData.DATA_SCHEMA, payload.toByteArray());
+ Optional result = ClusterEventTypeRegistry.decode(event);
+
+ assertTrue(result.isPresent());
+ RequestErrorEventData decoded = assertInstanceOf(RequestErrorEventData.class, result.get());
+ assertEquals(0, decoded.apiKey());
+ assertEquals(29, decoded.errorCode());
+ assertEquals(42.0, decoded.rps(), 0.001);
+ }
+
+ @Test
+ void decodeOffsetCommitFrequencyEvent() {
+ OffsetCommitFrequencyEvent payload = OffsetCommitFrequencyEvent.newBuilder()
+ .setGroupId("my-group")
+ .setTopic("my-topic")
+ .setRps(5000.0)
+ .build();
+
+ CloudEvent event = buildEvent(OffsetCommitFrequencyEventData.DATA_SCHEMA, payload.toByteArray());
+ Optional result = ClusterEventTypeRegistry.decode(event);
+
+ assertTrue(result.isPresent());
+ OffsetCommitFrequencyEventData decoded = assertInstanceOf(OffsetCommitFrequencyEventData.class, result.get());
+ assertEquals("my-group", decoded.groupId());
+ assertEquals(5000.0, decoded.rps(), 0.001);
+ }
+
+ @Test
+ void returnsEmptyForUnknownSchema() {
+ CloudEvent event = buildEvent("com.example.UnknownEvent", new byte[]{1, 2, 3});
+ assertFalse(ClusterEventTypeRegistry.decode(event).isPresent());
+ }
+
+ @Test
+ void returnsEmptyForNullDataSchema() {
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("test-id")
+ .withType("com.automq.ops.failover")
+ .withSource(URI.create("/automq/broker/0"))
+ .withData("application/protobuf", new byte[]{1})
+ .build();
+ assertFalse(ClusterEventTypeRegistry.decode(event).isPresent());
+ }
+
+ @Test
+ void returnsEmptyForCorruptPayload() {
+ CloudEvent event = buildEvent(FailoverEventData.DATA_SCHEMA, new byte[]{(byte) 0xFF, (byte) 0xFF});
+ assertFalse(ClusterEventTypeRegistry.decode(event).isPresent());
+ }
+
+ private static CloudEvent buildEvent(String dataSchema, byte[] payload) {
+ return CloudEventBuilder.v1()
+ .withId("test-id")
+ .withType("com.automq.test")
+ .withSource(URI.create("/automq/broker/0"))
+ .withDataSchema(URI.create(dataSchema))
+ .withData("application/protobuf", payload)
+ .build();
+ }
+}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
index ea3be5e650..9c180a90ec 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
@@ -1465,6 +1465,11 @@ public ExportClusterManifestResult exportClusterManifest(ExportClusterManifestOp
throw new UnsupportedOperationException();
}
+ @Override
+ public ClusterEventsReader describeClusterEvents(Long sinceMs) {
+ throw new UnsupportedOperationException();
+ }
+
// AutoMQ inject end
}
diff --git a/core/src/main/java/kafka/server/RequestErrorAccumulator.java b/core/src/main/java/kafka/server/RequestErrorAccumulator.java
new file mode 100644
index 0000000000..a043b81b9f
--- /dev/null
+++ b/core/src/main/java/kafka/server/RequestErrorAccumulator.java
@@ -0,0 +1,135 @@
+/*
+ * 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.server;
+
+import org.apache.kafka.clients.admin.ClusterEventPublisher;
+import org.apache.kafka.clients.admin.RequestErrorEventData;
+import org.apache.kafka.common.protocol.ApiKeys;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+
+public class RequestErrorAccumulator implements AutoCloseable {
+
+ private static final Logger log = LoggerFactory.getLogger(RequestErrorAccumulator.class);
+
+ private static final int MAX_CLIENTS = 10;
+
+ record ErrorKey(short apiKey, short errorCode, String resource) { }
+
+ static class ErrorBucket {
+ final LongAdder count = new LongAdder();
+ final Set clientIps = ConcurrentHashMap.newKeySet();
+ final Set clientIds = ConcurrentHashMap.newKeySet();
+ }
+
+ private final ConcurrentHashMap buckets = new ConcurrentHashMap<>();
+ private final int brokerId;
+ private final long flushIntervalMs;
+ private final ScheduledExecutorService scheduler;
+
+ public RequestErrorAccumulator(int brokerId, long flushIntervalMs) {
+ this.brokerId = brokerId;
+ this.flushIntervalMs = flushIntervalMs;
+ this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "request-error-accumulator-flush");
+ t.setDaemon(true);
+ return t;
+ });
+ this.scheduler.scheduleAtFixedRate(this::flush,
+ flushIntervalMs, flushIntervalMs, TimeUnit.MILLISECONDS);
+ }
+
+ public static boolean isRecordable(short errorCode) {
+ return errorCode != 0;
+ }
+
+ public void record(short apiKey, short errorCode, String resource,
+ String clientIp, String clientId) {
+ ErrorKey key = new ErrorKey(apiKey, errorCode, resource);
+ ErrorBucket bucket = buckets.computeIfAbsent(key, k -> new ErrorBucket());
+ bucket.count.increment();
+ if (bucket.clientIps.size() < MAX_CLIENTS) {
+ bucket.clientIps.add(clientIp);
+ }
+ if (bucket.clientIds.size() < MAX_CLIENTS) {
+ bucket.clientIds.add(clientId);
+ }
+ }
+
+ void flush() {
+ try {
+ doFlush();
+ } catch (Throwable e) {
+ log.warn("Error flushing request error events", e);
+ }
+ }
+
+ private void doFlush() {
+ double intervalSec = flushIntervalMs / 1000.0;
+ Iterator> it = buckets.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry entry = it.next();
+ // Remove the entry atomically before reading its values so that any new records
+ // arriving after this point go into a fresh bucket and are not silently dropped.
+ it.remove();
+ ErrorKey key = entry.getKey();
+ ErrorBucket bucket = entry.getValue();
+ long count = bucket.count.sum();
+ if (count == 0) {
+ continue;
+ }
+
+ String apiName = ApiKeys.forId(key.apiKey()).name;
+
+ RequestErrorEventData data = new RequestErrorEventData()
+ .setApiKey(key.apiKey())
+ .setErrorCode(key.errorCode())
+ .setClientIps(List.copyOf(bucket.clientIps))
+ .setClientIds(List.copyOf(bucket.clientIds))
+ .setRps(count / intervalSec);
+ if (!key.resource().isEmpty()) {
+ data.setResource(key.resource());
+ }
+
+ String subject = key.resource().isEmpty()
+ ? apiName : apiName + ":" + key.resource();
+
+ ClusterEventPublisher.publish(
+ RequestErrorEventData.TYPE,
+ "/automq/broker/" + brokerId,
+ subject,
+ RequestErrorEventData.DATA_SCHEMA,
+ data.toByteArray());
+ }
+ }
+
+ @Override
+ public void close() {
+ scheduler.shutdownNow();
+ }
+}
diff --git a/core/src/main/java/kafka/server/ResourceErrorExtractor.java b/core/src/main/java/kafka/server/ResourceErrorExtractor.java
new file mode 100644
index 0000000000..ba39e554aa
--- /dev/null
+++ b/core/src/main/java/kafka/server/ResourceErrorExtractor.java
@@ -0,0 +1,408 @@
+/*
+ * 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.server;
+
+import kafka.network.RequestChannel;
+
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.AddOffsetsToTxnRequest;
+import org.apache.kafka.common.requests.AddPartitionsToTxnResponse;
+import org.apache.kafka.common.requests.AlterConfigsResponse;
+import org.apache.kafka.common.requests.ConsumerGroupDescribeRequest;
+import org.apache.kafka.common.requests.ConsumerGroupHeartbeatRequest;
+import org.apache.kafka.common.requests.CreatePartitionsResponse;
+import org.apache.kafka.common.requests.CreateTopicsResponse;
+import org.apache.kafka.common.requests.DeleteGroupsResponse;
+import org.apache.kafka.common.requests.DeleteTopicsResponse;
+import org.apache.kafka.common.requests.DescribeConfigsResponse;
+import org.apache.kafka.common.requests.DescribeGroupsResponse;
+import org.apache.kafka.common.requests.EndTxnRequest;
+import org.apache.kafka.common.requests.FetchResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.FindCoordinatorResponse;
+import org.apache.kafka.common.requests.HeartbeatRequest;
+import org.apache.kafka.common.requests.IncrementalAlterConfigsResponse;
+import org.apache.kafka.common.requests.InitProducerIdRequest;
+import org.apache.kafka.common.requests.JoinGroupRequest;
+import org.apache.kafka.common.requests.LeaveGroupRequest;
+import org.apache.kafka.common.requests.ListOffsetsResponse;
+import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.requests.OffsetCommitRequest;
+import org.apache.kafka.common.requests.OffsetCommitResponse;
+import org.apache.kafka.common.requests.OffsetFetchResponse;
+import org.apache.kafka.common.requests.ProduceResponse;
+import org.apache.kafka.common.requests.SyncGroupRequest;
+import org.apache.kafka.common.requests.TxnOffsetCommitResponse;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Extracts (errorCode, resource) pairs from request/response for recording in RequestErrorAccumulator.
+ */
+public class ResourceErrorExtractor {
+
+ public record ResourceError(short errorCode, String resource) { }
+
+ /**
+ * Extract recordable errors with their associated resource from a response.
+ */
+ public static List extract(RequestChannel.Request request, AbstractResponse response) {
+ try {
+ return doExtract(request, response);
+ } catch (Exception e) {
+ // Never let extraction failures break request processing
+ return Collections.emptyList();
+ }
+ }
+
+ /**
+ * Best-effort resource extraction from request only (for closeConnection path).
+ */
+ public static String extractResourceFromRequest(RequestChannel.Request request) {
+ try {
+ return doExtractResourceFromRequest(request);
+ } catch (Exception e) {
+ return "";
+ }
+ }
+
+ private static List doExtract(RequestChannel.Request request, AbstractResponse response) {
+ ApiKeys apiKey = request.header().apiKey();
+ switch (apiKey) {
+ case PRODUCE:
+ return extractProduce((ProduceResponse) response);
+ case FETCH:
+ return extractFetch((FetchResponse) response);
+ case LIST_OFFSETS:
+ return extractListOffsets((ListOffsetsResponse) response);
+ case METADATA:
+ return extractMetadata((MetadataResponse) response);
+ case OFFSET_COMMIT:
+ return extractOffsetCommit(request, (OffsetCommitResponse) response);
+ case OFFSET_FETCH:
+ return extractOffsetFetch((OffsetFetchResponse) response);
+ case FIND_COORDINATOR:
+ return extractFindCoordinator(request, (FindCoordinatorResponse) response);
+ case CREATE_TOPICS:
+ return extractCreateTopics((CreateTopicsResponse) response);
+ case DELETE_TOPICS:
+ return extractDeleteTopics((DeleteTopicsResponse) response);
+ case CREATE_PARTITIONS:
+ return extractCreatePartitions((CreatePartitionsResponse) response);
+ default:
+ return doExtractOther(request, response, apiKey);
+ }
+ }
+
+ private static List doExtractOther(RequestChannel.Request request, AbstractResponse response, ApiKeys apiKey) {
+ switch (apiKey) {
+ case JOIN_GROUP:
+ return extractSingleErrorFromRequest(response, () -> request.body(JoinGroupRequest.class).data().groupId());
+ case SYNC_GROUP:
+ return extractSingleErrorFromRequest(response, () -> request.body(SyncGroupRequest.class).data().groupId());
+ case HEARTBEAT:
+ return extractSingleErrorFromRequest(response, () -> request.body(HeartbeatRequest.class).data().groupId());
+ case LEAVE_GROUP:
+ return extractSingleErrorFromRequest(response, () -> request.body(LeaveGroupRequest.class).data().groupId());
+ case DESCRIBE_GROUPS:
+ return extractDescribeGroups((DescribeGroupsResponse) response);
+ case DELETE_GROUPS:
+ return extractDeleteGroups((DeleteGroupsResponse) response);
+ case LIST_GROUPS:
+ return extractSingleError(response, "cluster");
+ case INIT_PRODUCER_ID:
+ return extractSingleErrorFromRequest(response, () -> {
+ String txnId = request.body(InitProducerIdRequest.class).data().transactionalId();
+ return txnId != null ? txnId : "";
+ });
+ case ADD_PARTITIONS_TO_TXN:
+ return extractAddPartitionsToTxn((AddPartitionsToTxnResponse) response);
+ case ADD_OFFSETS_TO_TXN:
+ return extractSingleErrorFromRequest(response, () -> request.body(AddOffsetsToTxnRequest.class).data().transactionalId());
+ case END_TXN:
+ return extractSingleErrorFromRequest(response, () -> request.body(EndTxnRequest.class).data().transactionalId());
+ case TXN_OFFSET_COMMIT:
+ return extractTxnOffsetCommit((TxnOffsetCommitResponse) response);
+ default:
+ return doExtractAdmin(request, response, apiKey);
+ }
+ }
+
+ private static List doExtractAdmin(RequestChannel.Request request, AbstractResponse response, ApiKeys apiKey) {
+ switch (apiKey) {
+ case ALTER_CONFIGS:
+ return extractAlterConfigs((AlterConfigsResponse) response);
+ case DESCRIBE_CONFIGS:
+ return extractDescribeConfigs((DescribeConfigsResponse) response);
+ case INCREMENTAL_ALTER_CONFIGS:
+ return extractIncrementalAlterConfigs((IncrementalAlterConfigsResponse) response);
+ case CREATE_DELEGATION_TOKEN:
+ case RENEW_DELEGATION_TOKEN:
+ case EXPIRE_DELEGATION_TOKEN:
+ return extractSingleError(response, "delegation-token");
+ case SASL_HANDSHAKE:
+ case SASL_AUTHENTICATE:
+ return extractSingleError(response, "sasl");
+ case CREATE_ACLS:
+ case DELETE_ACLS:
+ case DESCRIBE_ACLS:
+ return extractSingleError(response, "cluster");
+ case CONSUMER_GROUP_HEARTBEAT:
+ return extractSingleErrorFromRequest(response, () -> request.body(ConsumerGroupHeartbeatRequest.class).data().groupId());
+ case CONSUMER_GROUP_DESCRIBE:
+ return extractSingleErrorFromRequest(response, () -> {
+ List ids = request.body(ConsumerGroupDescribeRequest.class).data().groupIds();
+ return ids.isEmpty() ? "" : ids.get(0);
+ });
+ default:
+ return extractDefaultErrors(response);
+ }
+ }
+
+ // ---- Per-topic response extractors ----
+
+ private static List extractProduce(ProduceResponse response) {
+ List result = new ArrayList<>();
+ response.data().responses().forEach(topic ->
+ topic.partitionResponses().forEach(p ->
+ maybeAdd(result, p.errorCode(), topic.name())));
+ return result;
+ }
+
+ private static List extractFetch(FetchResponse response) {
+ List result = new ArrayList<>();
+ // top-level error
+ maybeAdd(result, response.error().code(), "");
+ response.data().responses().forEach(topic ->
+ topic.partitions().forEach(p ->
+ maybeAdd(result, p.errorCode(), topic.topic())));
+ return result;
+ }
+
+ private static List extractListOffsets(ListOffsetsResponse response) {
+ List result = new ArrayList<>();
+ response.data().topics().forEach(topic ->
+ topic.partitions().forEach(p ->
+ maybeAdd(result, p.errorCode(), topic.name())));
+ return result;
+ }
+
+ private static List extractMetadata(MetadataResponse response) {
+ List result = new ArrayList<>();
+ response.data().topics().forEach(topic ->
+ maybeAdd(result, topic.errorCode(), topic.name() != null ? topic.name() : ""));
+ return result;
+ }
+
+ private static List extractOffsetCommit(RequestChannel.Request request, OffsetCommitResponse response) {
+ List result = new ArrayList<>();
+ String groupId = request.body(OffsetCommitRequest.class).data().groupId();
+ response.data().topics().forEach(topic ->
+ topic.partitions().forEach(p -> {
+ short code = p.errorCode();
+ if (code == Errors.GROUP_AUTHORIZATION_FAILED.code()) {
+ maybeAdd(result, code, groupId);
+ } else {
+ maybeAdd(result, code, topic.name());
+ }
+ }));
+ return result;
+ }
+
+ private static List extractOffsetFetch(OffsetFetchResponse response) {
+ List result = new ArrayList<>();
+ // v8+ batched groups
+ response.data().groups().forEach(group ->
+ maybeAdd(result, group.errorCode(), group.groupId()));
+ // older single-group
+ if (response.data().groups().isEmpty()) {
+ maybeAdd(result, response.data().errorCode(), "");
+ }
+ return result;
+ }
+
+ private static List extractFindCoordinator(RequestChannel.Request request, FindCoordinatorResponse response) {
+ List result = new ArrayList<>();
+ if (!response.data().coordinators().isEmpty()) {
+ response.data().coordinators().forEach(c ->
+ maybeAdd(result, c.errorCode(), c.key()));
+ } else {
+ String key = request.body(FindCoordinatorRequest.class).data().key();
+ maybeAdd(result, response.data().errorCode(), key != null ? key : "");
+ }
+ return result;
+ }
+
+ private static List extractCreateTopics(CreateTopicsResponse response) {
+ List result = new ArrayList<>();
+ response.data().topics().forEach(topic ->
+ maybeAdd(result, topic.errorCode(), topic.name()));
+ return result;
+ }
+
+ private static List extractDeleteTopics(DeleteTopicsResponse response) {
+ List result = new ArrayList<>();
+ response.data().responses().forEach(topic ->
+ maybeAdd(result, topic.errorCode(), topic.name() != null ? topic.name() : ""));
+ return result;
+ }
+
+ private static List extractCreatePartitions(CreatePartitionsResponse response) {
+ List result = new ArrayList<>();
+ response.data().results().forEach(r ->
+ maybeAdd(result, r.errorCode(), r.name()));
+ return result;
+ }
+
+ // ---- Per-group response extractors ----
+
+ private static List extractDescribeGroups(DescribeGroupsResponse response) {
+ List result = new ArrayList<>();
+ response.data().groups().forEach(g ->
+ maybeAdd(result, g.errorCode(), g.groupId()));
+ return result;
+ }
+
+ private static List extractDeleteGroups(DeleteGroupsResponse response) {
+ List result = new ArrayList<>();
+ response.data().results().forEach(r ->
+ maybeAdd(result, r.errorCode(), r.groupId()));
+ return result;
+ }
+
+ // ---- Transaction extractors ----
+
+ private static List extractAddPartitionsToTxn(AddPartitionsToTxnResponse response) {
+ List result = new ArrayList<>();
+ // v3 and below: per-topic results
+ response.data().resultsByTopicV3AndBelow().forEach(topic ->
+ topic.resultsByPartition().forEach(p ->
+ maybeAdd(result, p.partitionErrorCode(), topic.name())));
+ // v4+: per-transaction, per-topic results
+ response.data().resultsByTransaction().forEach(txn ->
+ txn.topicResults().forEach(topic ->
+ topic.resultsByPartition().forEach(p ->
+ maybeAdd(result, p.partitionErrorCode(), topic.name()))));
+ return result;
+ }
+
+ private static List extractTxnOffsetCommit(TxnOffsetCommitResponse response) {
+ List result = new ArrayList<>();
+ response.data().topics().forEach(topic ->
+ topic.partitions().forEach(p ->
+ maybeAdd(result, p.errorCode(), topic.name())));
+ return result;
+ }
+
+ // ---- Config extractors ----
+
+ private static List extractAlterConfigs(AlterConfigsResponse response) {
+ List result = new ArrayList<>();
+ response.data().responses().forEach(r ->
+ maybeAdd(result, r.errorCode(), r.resourceName()));
+ return result;
+ }
+
+ private static List extractDescribeConfigs(DescribeConfigsResponse response) {
+ List result = new ArrayList<>();
+ response.data().results().forEach(r ->
+ maybeAdd(result, r.errorCode(), r.resourceName()));
+ return result;
+ }
+
+ private static List extractIncrementalAlterConfigs(IncrementalAlterConfigsResponse response) {
+ List result = new ArrayList<>();
+ response.data().responses().forEach(r ->
+ maybeAdd(result, r.errorCode(), r.resourceName()));
+ return result;
+ }
+
+ // ---- Helpers ----
+
+ private static List extractSingleError(AbstractResponse response, String resource) {
+ List result = new ArrayList<>();
+ response.errorCounts().forEach((error, count) ->
+ maybeAdd(result, error.code(), resource));
+ return result;
+ }
+
+ @FunctionalInterface
+ private interface ResourceSupplier {
+ String get();
+ }
+
+ private static List extractSingleErrorFromRequest(
+ AbstractResponse response, ResourceSupplier resourceSupplier) {
+ List result = new ArrayList<>();
+ String resource = resourceSupplier.get();
+ response.errorCounts().forEach((error, count) ->
+ maybeAdd(result, error.code(), resource));
+ return result;
+ }
+
+ private static List extractDefaultErrors(AbstractResponse response) {
+ List result = new ArrayList<>();
+ response.errorCounts().forEach((error, count) ->
+ maybeAdd(result, error.code(), ""));
+ return result;
+ }
+
+ private static void maybeAdd(List result, short errorCode, String resource) {
+ if (errorCode != 0) {
+ result.add(new ResourceError(errorCode, resource));
+ }
+ }
+
+ private static String doExtractResourceFromRequest(RequestChannel.Request request) {
+ ApiKeys apiKey = request.header().apiKey();
+ switch (apiKey) {
+ case PRODUCE:
+ var produceTopics = request.body(org.apache.kafka.common.requests.ProduceRequest.class).data().topicData();
+ return produceTopics.isEmpty() ? "" : produceTopics.iterator().next().name();
+ case JOIN_GROUP:
+ return request.body(JoinGroupRequest.class).data().groupId();
+ case SYNC_GROUP:
+ return request.body(SyncGroupRequest.class).data().groupId();
+ case HEARTBEAT:
+ return request.body(HeartbeatRequest.class).data().groupId();
+ case LEAVE_GROUP:
+ return request.body(LeaveGroupRequest.class).data().groupId();
+ case OFFSET_COMMIT:
+ return request.body(OffsetCommitRequest.class).data().groupId();
+ case INIT_PRODUCER_ID: {
+ String txnId = request.body(InitProducerIdRequest.class).data().transactionalId();
+ return txnId != null ? txnId : "";
+ }
+ case ADD_OFFSETS_TO_TXN:
+ return request.body(AddOffsetsToTxnRequest.class).data().transactionalId();
+ case END_TXN:
+ return request.body(EndTxnRequest.class).data().transactionalId();
+ case FIND_COORDINATOR: {
+ String key = request.body(FindCoordinatorRequest.class).data().key();
+ return key != null ? key : "";
+ }
+ case CONSUMER_GROUP_HEARTBEAT:
+ return request.body(ConsumerGroupHeartbeatRequest.class).data().groupId();
+ default:
+ return "";
+ }
+ }
+}
diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala
index ac856f5e27..a9ce32e09e 100644
--- a/core/src/main/scala/kafka/network/RequestChannel.scala
+++ b/core/src/main/scala/kafka/network/RequestChannel.scala
@@ -21,7 +21,7 @@ import com.fasterxml.jackson.databind.JsonNode
import com.typesafe.scalalogging.Logger
import com.yammer.metrics.core.{Histogram, Meter}
import kafka.network
-import kafka.server.{KafkaConfig, RequestLocal}
+import kafka.server.{KafkaConfig, RequestErrorAccumulator, RequestLocal, ResourceErrorExtractor}
import kafka.utils.Implicits._
import kafka.utils.{Logging, Pool}
import org.apache.kafka.common.config.ConfigResource
@@ -188,6 +188,12 @@ object RequestChannel extends Logging {
}
}
+ // AutoMQ inject start
+ def body[T <: AbstractRequest](clazz: Class[T]): T = {
+ clazz.cast(bodyAndSize.request)
+ }
+ // AutoMQ inject end
+
def loggableRequest: AbstractRequest = {
bodyAndSize.request match {
@@ -392,6 +398,10 @@ class RequestChannel(val queueSize: Int,
private val multiCallbackQueue = new java.util.ArrayList[ArrayBlockingQueue[BaseRequest]]()
private var notifiedShutdown = false
+ // AutoMQ inject start - request error accumulator
+ @volatile var requestErrorAccumulator: RequestErrorAccumulator = _
+ // AutoMQ inject end
+
metricsGroup.newGauge(requestQueueSizeMetricName, () => {
if (multiRequestQueue.size() != 0) {
multiRequestQueue.stream().mapToInt(q => q.size()).sum()
@@ -456,6 +466,9 @@ class RequestChannel(val queueSize: Int,
// This case is used when the request handler has encountered an error, but the client
// does not expect a response (e.g. when produce request has acks set to 0)
updateErrorMetrics(request.header.apiKey, errorCounts.asScala)
+ // AutoMQ inject start
+ maybeRecordRequestErrorsFromMap(request, errorCounts)
+ // AutoMQ inject end
sendResponse(new RequestChannel.CloseConnectionResponse(request))
}
@@ -465,6 +478,9 @@ class RequestChannel(val queueSize: Int,
onComplete: Option[Send => Unit]
): Unit = {
updateErrorMetrics(request.header.apiKey, response.errorCounts.asScala)
+ // AutoMQ inject start
+ maybeRecordRequestErrors(request, response)
+ // AutoMQ inject end
sendResponse(new RequestChannel.SendResponse(
request,
request.buildResponseSend(response),
@@ -568,6 +584,40 @@ class RequestChannel(val queueSize: Int,
def receiveRequest(): RequestChannel.BaseRequest =
requestQueue.take()
+ // AutoMQ inject start
+ private def maybeRecordRequestErrors(
+ request: RequestChannel.Request,
+ response: AbstractResponse
+ ): Unit = {
+ val acc = requestErrorAccumulator
+ if (acc == null) return
+ val errors = ResourceErrorExtractor.extract(request, response)
+ if (errors.isEmpty) return
+ val clientIp = request.context.clientAddress.getHostAddress
+ val clientId = request.header.clientId
+ val apiKey = request.header.apiKey.id
+ errors.forEach { re =>
+ acc.record(apiKey, re.errorCode, re.resource, clientIp, clientId)
+ }
+ }
+
+ private def maybeRecordRequestErrorsFromMap(
+ request: RequestChannel.Request,
+ errorCounts: java.util.Map[Errors, Integer]
+ ): Unit = {
+ val acc = requestErrorAccumulator
+ if (acc == null) return
+ val clientIp = request.context.clientAddress.getHostAddress
+ val clientId = request.header.clientId
+ val apiKey = request.header.apiKey.id
+ val resource = ResourceErrorExtractor.extractResourceFromRequest(request)
+ errorCounts.forEach { (error, _) =>
+ if (RequestErrorAccumulator.isRecordable(error.code))
+ acc.record(apiKey, error.code, resource, clientIp, clientId)
+ }
+ }
+ // AutoMQ inject end
+
def updateErrorMetrics(apiKey: ApiKeys, errors: collection.Map[Errors, Integer]): Unit = {
errors.forKeyValue { (error, count) =>
metrics(apiKey.name).markErrorMeter(error, count)
diff --git a/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala b/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala
index 9060c464bf..c217d342c6 100644
--- a/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala
+++ b/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala
@@ -24,10 +24,11 @@ import kafka.controller.KafkaController
import kafka.coordinator.transaction.TransactionCoordinator
import kafka.utils.Logging
import org.apache.kafka.clients.ClientResponse
+import org.apache.kafka.clients.admin.ClusterEventsConfig
import org.apache.kafka.common.config.TopicConfig
import org.apache.kafka.common.errors.InvalidTopicException
import org.apache.kafka.common.internals.Topic
-import org.apache.kafka.common.internals.Topic.{GROUP_METADATA_TOPIC_NAME, TABLE_TOPIC_CONTROL_TOPIC_NAME, TABLE_TOPIC_DATA_TOPIC_NAME, TRANSACTION_STATE_TOPIC_NAME}
+import org.apache.kafka.common.internals.Topic.{CLUSTER_EVENTS_TOPIC_NAME, GROUP_METADATA_TOPIC_NAME, TABLE_TOPIC_CONTROL_TOPIC_NAME, TABLE_TOPIC_DATA_TOPIC_NAME, TRANSACTION_STATE_TOPIC_NAME}
import org.apache.kafka.common.message.CreateTopicsRequestData
import org.apache.kafka.common.message.CreateTopicsRequestData.{CreatableTopic, CreatableTopicConfig, CreatableTopicConfigCollection}
import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopic
@@ -265,6 +266,15 @@ class DefaultAutoTopicCreationManager(
.setReplicationFactor(1)
.setConfigs(convertToTopicConfigCollections(configs))
}
+ case CLUSTER_EVENTS_TOPIC_NAME => {
+ val configs = new Properties()
+ configs.put(TopicConfig.RETENTION_MS_CONFIG, ClusterEventsConfig.DEFAULT_RETENTION_MS)
+ new CreatableTopic()
+ .setName(topic)
+ .setNumPartitions(ClusterEventsConfig.DEFAULT_PARTITIONS)
+ .setReplicationFactor(1)
+ .setConfigs(convertToTopicConfigCollections(configs))
+ }
// AutoMQ inject end
case topicName =>
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
index 28ad477b4b..38fda51d98 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -174,6 +174,7 @@ class BrokerServer(
config.addReconfigurable(clientRackProvider)
var tableManager: TableManager = _
+ var requestErrorAccumulator: RequestErrorAccumulator = _
// AutoMQ inject end
private def maybeChangeStatus(from: ProcessStatus, to: ProcessStatus): Boolean = {
@@ -462,6 +463,22 @@ class BrokerServer(
config.numIoThreads, s"${DataPlaneAcceptor.MetricPrefix}RequestHandlerAvgIdlePercent",
DataPlaneAcceptor.ThreadPrefix)
+ // AutoMQ inject start - cluster event publisher and request error accumulator
+ try {
+ val publisherConfig = new java.util.HashMap[String, Object]()
+ kafka.automq.utils.ClientUtils.clusterClientBaseConfig(config).forEach { (k, v) =>
+ publisherConfig.put(k.toString, v)
+ }
+ config.rack.foreach(rack => publisherConfig.put("client.id", "automq_az=" + rack))
+ org.apache.kafka.clients.admin.ClusterEventPublisher.setup(publisherConfig)
+ requestErrorAccumulator = new RequestErrorAccumulator(config.nodeId, 30000L)
+ socketServer.dataPlaneRequestChannel.requestErrorAccumulator = requestErrorAccumulator
+ } catch {
+ case e: Throwable =>
+ error("Failed to initialize ClusterEventPublisher/RequestErrorAccumulator", e)
+ }
+ // AutoMQ inject end
+
// Start RemoteLogManager before initializing broker metadata publishers.
remoteLogManagerOpt.foreach { rlm =>
val listenerName = config.remoteLogManagerConfig.remoteLogMetadataManagerListenerName()
@@ -797,6 +814,12 @@ class BrokerServer(
if (quotaManagers != null)
CoreUtils.swallow(quotaManagers.shutdown(), this)
+ // AutoMQ inject start - shutdown request error accumulator and cluster event publisher
+ if (requestErrorAccumulator != null)
+ CoreUtils.swallow(requestErrorAccumulator.close(), this)
+ CoreUtils.swallow(org.apache.kafka.clients.admin.ClusterEventPublisher.shutdown(), this)
+ // AutoMQ inject end
+
if (socketServer != null)
CoreUtils.swallow(socketServer.shutdown(), this)
if (brokerTopicStats != null)
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index 5ac924ab36..6967f95649 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -1321,10 +1321,10 @@ class KafkaApis(val requestChannel: RequestChannel,
autoTopicCreationManager.createTopics(nonExistingTopics, controllerMutationQuota, Some(request.context))
} else {
// AutoMQ inject start
- for (tableTopic <- Set(Topic.TABLE_TOPIC_CONTROL_TOPIC_NAME, Topic.TABLE_TOPIC_DATA_TOPIC_NAME)) {
- if (nonExistingTopics.contains(tableTopic)) {
+ for (autoCreateTopic <- Set(Topic.TABLE_TOPIC_CONTROL_TOPIC_NAME, Topic.TABLE_TOPIC_DATA_TOPIC_NAME, Topic.CLUSTER_EVENTS_TOPIC_NAME)) {
+ if (nonExistingTopics.contains(autoCreateTopic)) {
val controllerMutationQuota = quotas.controllerMutation.newPermissiveQuotaFor(request)
- autoTopicCreationManager.createTopics(Set(tableTopic), controllerMutationQuota, Some(request.context))
+ autoTopicCreationManager.createTopics(Set(autoCreateTopic), controllerMutationQuota, Some(request.context))
}
}
// AutoMQ inject end
diff --git a/core/src/test/java/kafka/server/RequestErrorEventDataTest.java b/core/src/test/java/kafka/server/RequestErrorEventDataTest.java
new file mode 100644
index 0000000000..839e431af7
--- /dev/null
+++ b/core/src/test/java/kafka/server/RequestErrorEventDataTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.server;
+
+import org.apache.kafka.clients.admin.RequestErrorEventData;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+class RequestErrorEventDataTest {
+
+ @Test
+ void roundTripSerialization() {
+ RequestErrorEventData original = new RequestErrorEventData()
+ .setApiKey(0)
+ .setErrorCode(29)
+ .setResource("my-topic")
+ .setClientIps(List.of("10.0.0.1"))
+ .setClientIds(List.of("producer-1"))
+ .setRps(42.0);
+
+ byte[] bytes = original.toByteArray();
+ assertNotNull(bytes);
+
+ RequestErrorEventData decoded = RequestErrorEventData.fromByteArray(bytes);
+ assertEquals(0, decoded.apiKey());
+ assertEquals(29, decoded.errorCode());
+ assertEquals("my-topic", decoded.resource());
+ assertEquals(List.of("10.0.0.1"), decoded.clientIps());
+ assertEquals(List.of("producer-1"), decoded.clientIds());
+ assertEquals(42.0, decoded.rps(), 0.001);
+ }
+}
diff --git a/devkit/Dockerfile b/devkit/Dockerfile
index 185913a01c..33fd495f46 100644
--- a/devkit/Dockerfile
+++ b/devkit/Dockerfile
@@ -2,11 +2,13 @@ ARG jdk_version=eclipse-temurin:17-jdk-jammy
FROM $jdk_version
ENV DEBIAN_FRONTEND=noninteractive
+ENV PATH="/root/.local/bin:$PATH"
RUN apt-get update && apt-get install -y \
- curl jq iproute2 iptables iputils-ping net-tools wget \
+ curl jq iproute2 iptables iputils-ping net-tools wget python3 \
&& wget -O /usr/local/bin/jmxterm.jar https://github.com/jiaqi/jmxterm/releases/download/v1.0.4/jmxterm-1.0.4-uber.jar \
&& mkdir -p /opt/arthas && wget -O /opt/arthas/arthas-boot.jar https://arthas.aliyun.com/arthas-boot.jar \
+ && curl -LsSf https://astral.sh/uv/install.sh | sh \
&& apt-get -y clean && rm -rf /var/lib/apt/lists/*
WORKDIR /opt/automq
diff --git a/devkit/README.md b/devkit/README.md
index 1ef8fb6f5e..20e6d3d587 100644
--- a/devkit/README.md
+++ b/devkit/README.md
@@ -313,6 +313,29 @@ just start 5 zerozone analytics # 5-node + zone router + query engines
| `telemetry` | OTel metrics export via OTLP HTTP — **edit `config/features/telemetry.properties` to set collector endpoint before use** | — (config only) |
| `analytics` | Spark + Trino for querying Iceberg tables | Spark (:8888), Trino (:8090) |
+## Namespace (Multi-Instance Isolation)
+
+Run multiple DevKit instances on the same machine without port conflicts. Each namespace gets a unique `COMPOSE_PROJECT_NAME` and port offset.
+
+**Requires separate git worktrees** — each worktree has its own `devkit/` directory and `.devkit/compose.env`, so each instance is fully isolated. You cannot run multiple namespaces from the same devkit directory.
+
+```bash
+# In worktree A:
+just set-ns lag-test 1 # ID=1 → ports offset by +100
+just start 3
+
+# In worktree B:
+just set-ns perf 2 # ID=2 → ports offset by +200
+just start
+
+# Revert to default ports
+just clear-ns
+```
+
+- Name: 1-8 chars, must start with a letter, only `[a-z0-9-]`
+- ID: 1-9, determines port offset (`ID * 100`)
+- Config is stored in `.devkit/compose.env` (gitignored)
+
## Port Allocation
| Node | Kafka | Controller | JDWP | JMX |
@@ -353,6 +376,8 @@ Other services:
| `S3_DATA_BUCKET` | `automq-data` | Data bucket |
| `S3_OPS_BUCKET` | `automq-ops` | Ops bucket |
| `S3_PATH_STYLE` | `true` | Path-style access (required for MinIO) |
+| `S3_ACCESS_KEY` | `admin` | S3 access key (exported as `KAFKA_S3_ACCESS_KEY` to containers) |
+| `S3_SECRET_KEY` | `password` | S3 secret key (exported as `KAFKA_S3_SECRET_KEY` to containers) |
**Per-node variables** (computed per node, Layer 2/3/4):
@@ -373,6 +398,8 @@ S3_ENDPOINT := "https://s3.ap-northeast-1.amazonaws.com"
S3_DATA_BUCKET := "my-automq-data"
S3_OPS_BUCKET := "my-automq-ops"
S3_PATH_STYLE := "false"
+S3_ACCESS_KEY := "AKIA..."
+S3_SECRET_KEY := "..."
```
**Change listeners (e.g. SASL)** — edit `config/role/server.properties`:
diff --git a/devkit/docker-compose.yml b/devkit/docker-compose.yml
index 74e18699cc..a5ea6bf629 100644
--- a/devkit/docker-compose.yml
+++ b/devkit/docker-compose.yml
@@ -12,10 +12,11 @@ x-automq-common: &automq-common
labels:
com.automq.devkit: "true"
environment:
- - KAFKA_S3_ACCESS_KEY=admin
- - KAFKA_S3_SECRET_KEY=password
+ - KAFKA_S3_ACCESS_KEY=${KAFKA_S3_ACCESS_KEY}
+ - KAFKA_S3_SECRET_KEY=${KAFKA_S3_SECRET_KEY}
- KAFKA_HEAP_OPTS=${KAFKA_HEAP_OPTS:--Xms1g -Xmx4g -XX:MetaspaceSize=96m -XX:MaxDirectMemorySize=1G}
- KAFKA_JVM_PERFORMANCE_OPTS=-server -XX:+UseZGC -XX:ZCollectionInterval=5
+ - LOG_DIR=/tmp/kafka-logs
depends_on:
mc:
condition: service_completed_successfully
@@ -35,15 +36,14 @@ services:
# ==================== Infrastructure ====================
minio:
image: minio/minio:RELEASE.2025-05-24T17-08-30Z
- container_name: minio
hostname: warehouse.minio
environment:
- MINIO_ROOT_USER=admin
- MINIO_ROOT_PASSWORD=password
- MINIO_DOMAIN=minio
ports:
- - "9000:9000"
- - "9001:9001"
+ - "${MINIO_API_PORT:-9000}:9000"
+ - "${MINIO_CONSOLE_PORT:-9001}:9001"
command: ["server", "/data", "--console-address", ":9001"]
healthcheck:
test: ["CMD", "curl", "-f", "http://minio:9000/minio/health/live"]
@@ -53,7 +53,6 @@ services:
mc:
image: minio/mc:RELEASE.2025-05-21T01-59-54Z
- container_name: mc
depends_on:
minio:
condition: service_healthy
@@ -69,13 +68,12 @@ services:
# ==================== AutoMQ Nodes ====================
node-0:
<<: *automq-common
- container_name: node-0
hostname: node-0
profiles: [single, cluster, cluster4, cluster5]
ports:
- - "9092:9192"
- - "5005:5005"
- - "9999:9999"
+ - "${NODE0_KAFKA_PORT:-9092}:9192"
+ - "${NODE0_DEBUG_PORT:-5005}:5005"
+ - "${NODE0_JMX_PORT:-9999}:9999"
command:
- bash
- -c
@@ -88,12 +86,11 @@ services:
node-1:
<<: *automq-common
- container_name: node-1
hostname: node-1
profiles: [cluster, cluster4, cluster5]
ports:
- - "19092:9192"
- - "5006:5005"
+ - "${NODE1_KAFKA_PORT:-19092}:9192"
+ - "${NODE1_DEBUG_PORT:-5006}:5005"
command:
- bash
- -c
@@ -106,12 +103,11 @@ services:
node-2:
<<: *automq-common
- container_name: node-2
hostname: node-2
profiles: [cluster, cluster4, cluster5]
ports:
- - "29092:9192"
- - "5007:5005"
+ - "${NODE2_KAFKA_PORT:-29092}:9192"
+ - "${NODE2_DEBUG_PORT:-5007}:5005"
command:
- bash
- -c
@@ -124,12 +120,11 @@ services:
node-3:
<<: *automq-common
- container_name: node-3
hostname: node-3
profiles: [cluster4, cluster5]
ports:
- - "39092:9192"
- - "5008:5005"
+ - "${NODE3_KAFKA_PORT:-39092}:9192"
+ - "${NODE3_DEBUG_PORT:-5008}:5005"
command:
- bash
- -c
@@ -142,12 +137,11 @@ services:
node-4:
<<: *automq-common
- container_name: node-4
hostname: node-4
profiles: [cluster5]
ports:
- - "49092:9192"
- - "5009:5005"
+ - "${NODE4_KAFKA_PORT:-49092}:9192"
+ - "${NODE4_DEBUG_PORT:-5009}:5005"
command:
- bash
- -c
@@ -161,11 +155,10 @@ services:
# ==================== TableTopic ====================
schema-registry:
image: confluentinc/cp-schema-registry:7.7.8
- container_name: schema-registry
hostname: schema-registry
profiles: [tabletopic]
ports:
- - "8081:8081"
+ - "${SCHEMA_REGISTRY_PORT:-8081}:8081"
environment:
SCHEMA_REGISTRY_HOST_NAME: schema-registry
SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: node-0:9092
@@ -179,11 +172,10 @@ services:
rest:
image: apache/iceberg-rest-fixture
- container_name: iceberg-rest
hostname: rest
profiles: [tabletopic]
ports:
- - "8181:8181"
+ - "${ICEBERG_REST_PORT:-8181}:8181"
depends_on:
minio:
condition: service_healthy
@@ -204,7 +196,6 @@ services:
# ==================== Analytics ====================
spark-iceberg:
image: tabulario/spark-iceberg:3.5.5_1.8.1
- container_name: spark
hostname: spark-iceberg
profiles: [analytics]
depends_on:
@@ -215,18 +206,17 @@ services:
- AWS_SECRET_ACCESS_KEY=password
- AWS_REGION=us-east-1
ports:
- - "8888:8888"
- - "8080:8080"
+ - "${SPARK_NOTEBOOK_PORT:-8888}:8888"
+ - "${SPARK_UI_PORT:-8080}:8080"
trino:
image: trinodb/trino:472
- container_name: trino
hostname: trino
profiles: [analytics]
depends_on:
minio:
condition: service_healthy
ports:
- - "8090:8080"
+ - "${TRINO_PORT:-8090}:8080"
volumes:
- ./config/trino-catalog:/etc/trino/catalog
diff --git a/devkit/justfile b/devkit/justfile
index 1b3604a1b5..dcf817a06d 100644
--- a/devkit/justfile
+++ b/devkit/justfile
@@ -3,20 +3,25 @@
set dotenv-load := false
export AUTOMQ_DEV_HOME := justfile_directory() + "/.."
+export KAFKA_S3_ACCESS_KEY := S3_ACCESS_KEY
+export KAFKA_S3_SECRET_KEY := S3_SECRET_KEY
DEVKIT := justfile_directory() + "/.devkit"
-COMPOSE := "docker compose -f " + justfile_directory() + "/docker-compose.yml"
+_ENV_FILE_FLAG := if path_exists(justfile_directory() + "/.devkit/compose.env") == "true" { "--env-file " + justfile_directory() + "/.devkit/compose.env" } else { "" }
+COMPOSE := "docker compose -f " + justfile_directory() + "/docker-compose.yml " + _ENV_FILE_FLAG
ALL_PROFILES := "--profile single --profile cluster --profile cluster4 --profile cluster5 --profile tabletopic --profile analytics"
AZ_NAMES := "az-0 az-1 az-2"
HEAP_OPTS := "-Xms256m -Xmx256m"
KAFKA_OPTS := "KAFKA_HEAP_OPTS='" + HEAP_OPTS + "' KAFKA_JVM_PERFORMANCE_OPTS='' JMX_PORT=''"
# S3 configuration
-S3_REGION := "us-east-1"
-S3_ENDPOINT := "http://minio:9000"
+S3_REGION := "us-east-1"
+S3_ENDPOINT := "http://minio:9000"
S3_DATA_BUCKET := "automq-data"
S3_OPS_BUCKET := "automq-ops"
S3_PATH_STYLE := "true"
+S3_ACCESS_KEY := "admin"
+S3_SECRET_KEY := "password"
default:
@just --list
@@ -81,7 +86,8 @@ _render-node node ctrl_count cluster_id voters bootstrap features="":
export VOTERS="{{voters}}"
export BOOTSTRAP="{{bootstrap}}"
export NODE_ID="{{node}}"
- export HOST_PORT=$(({{node}} * 10000 + 9092))
+ PORT_OFFSET=$(grep -s '^PORT_OFFSET=' "{{DEVKIT}}/compose.env" | cut -d= -f2)
+ export HOST_PORT=$(({{node}} * 10000 + 9092 + ${PORT_OFFSET:-0}))
export AZ_NAME="${AZS[$(({{node}} % ${#AZS[@]}))]}"
render() { envsubst < "$1" | grep -v '^#' | grep -v '^$'; }
@@ -123,16 +129,83 @@ generate-config nodes features="":
[private]
_node-list:
- @docker ps --filter "label=com.automq.devkit=true" --format '{{ '{{' }}.Names{{ '}}' }}'
+ @{{COMPOSE}} ps --format '{{ '{{' }}.Service{{ '}}' }}' 2>/dev/null | grep 'node-' || true
[private]
_ensure-iptables-chain node:
#!/usr/bin/env bash
- docker exec {{node}} iptables -N DEVKIT 2>/dev/null || true
- docker exec {{node}} iptables -C OUTPUT -j DEVKIT 2>/dev/null || \
- docker exec {{node}} iptables -A OUTPUT -j DEVKIT
- docker exec {{node}} iptables -C INPUT -j DEVKIT 2>/dev/null || \
- docker exec {{node}} iptables -A INPUT -j DEVKIT
+ {{COMPOSE}} exec {{node}} iptables -N DEVKIT 2>/dev/null || true
+ {{COMPOSE}} exec {{node}} iptables -C OUTPUT -j DEVKIT 2>/dev/null || \
+ {{COMPOSE}} exec {{node}} iptables -A OUTPUT -j DEVKIT
+ {{COMPOSE}} exec {{node}} iptables -C INPUT -j DEVKIT 2>/dev/null || \
+ {{COMPOSE}} exec {{node}} iptables -A INPUT -j DEVKIT
+
+# ==================== Namespace ====================
+
+[doc("Set namespace for multi-instance isolation. Example: just set-ns lag-test 1")]
+set-ns name id:
+ #!/usr/bin/env bash
+ set -e
+ NAME="{{name}}"
+ ID="{{id}}"
+ if ! [[ "$NAME" =~ ^[a-z][a-z0-9-]{0,7}$ ]]; then
+ echo "✗ Name must be 1-8 chars, start with a letter, only [a-z0-9-]"
+ exit 1
+ fi
+ if ! [[ "$ID" =~ ^[1-9]$ ]]; then
+ echo "✗ ID must be 1-9"
+ exit 1
+ fi
+
+ OFFSET=$((ID * 100))
+ CHECK_PORT=$((9092 + OFFSET))
+
+ if lsof -i ":${CHECK_PORT}" >/dev/null 2>&1; then
+ echo "✗ Port ${CHECK_PORT} is already in use. Try a different ID."
+ exit 1
+ fi
+
+ # Base ports — each gets +OFFSET in the generated env file
+ declare -a PORT_DEFS=(
+ "NODE0_KAFKA_PORT=9092"
+ "NODE0_DEBUG_PORT=5005"
+ "NODE0_JMX_PORT=9999"
+ "NODE1_KAFKA_PORT=19092"
+ "NODE1_DEBUG_PORT=5006"
+ "NODE2_KAFKA_PORT=29092"
+ "NODE2_DEBUG_PORT=5007"
+ "NODE3_KAFKA_PORT=39092"
+ "NODE3_DEBUG_PORT=5008"
+ "NODE4_KAFKA_PORT=49092"
+ "NODE4_DEBUG_PORT=5009"
+ "MINIO_API_PORT=9000"
+ "MINIO_CONSOLE_PORT=9001"
+ "SCHEMA_REGISTRY_PORT=8081"
+ "ICEBERG_REST_PORT=8181"
+ "SPARK_NOTEBOOK_PORT=8888"
+ "SPARK_UI_PORT=8080"
+ "TRINO_PORT=8090"
+ )
+
+ mkdir -p "{{DEVKIT}}"
+ {
+ echo "COMPOSE_PROJECT_NAME=devkit-$NAME"
+ echo "PORT_OFFSET=$OFFSET"
+ for def in "${PORT_DEFS[@]}"; do
+ key="${def%%=*}"
+ base="${def#*=}"
+ echo "${key}=$((base + OFFSET))"
+ done
+ } > "{{DEVKIT}}/compose.env"
+
+ echo "✓ Namespace: ${NAME} (id=${ID})"
+ echo " Port offset: +${OFFSET}"
+ echo " Kafka will be on port ${CHECK_PORT}"
+
+[doc("Clear namespace, revert to default ports")]
+clear-ns:
+ rm -f "{{DEVKIT}}/compose.env"
+ @echo "✓ Namespace cleared — using default ports"
# ==================== Lifecycle ====================
@@ -152,10 +225,12 @@ start nodes="1" *FEATURES: ensure-cluster-id
echo "✓ Starting {{nodes}} node(s)..."
{{COMPOSE}} $COMPOSE_PROFILES up -d
echo ""
+ PORT_OFFSET=$(grep -s '^PORT_OFFSET=' "{{DEVKIT}}/compose.env" | cut -d= -f2)
+ PORT_OFFSET=${PORT_OFFSET:-0}
echo "🎉 AutoMQ DevKit started"
- echo " Kafka: localhost:9092"
- echo " Debug: localhost:5005"
- echo " MinIO: http://localhost:9001 (admin/password)"
+ echo " Kafka: localhost:$((9092 + PORT_OFFSET))"
+ echo " Debug: localhost:$((5005 + PORT_OFFSET))"
+ echo " MinIO: http://localhost:$((9001 + PORT_OFFSET)) (admin/password)"
just wait
[doc("Build image & code, then start. Example: just start-build 3 tabletopic")]
@@ -202,11 +277,11 @@ logs-follow node="":
[doc("Enter container shell (default: node 0)")]
shell node="0":
- @docker exec -it node-{{node}} bash
+ @{{COMPOSE}} exec node-{{node}} bash
[doc("Run a single command in container. Example: just exec 0 ls /opt/automq/bin")]
exec node *CMD:
- @docker exec node-{{node}} {{CMD}}
+ @{{COMPOSE}} exec node-{{node}} {{CMD}}
[doc("Wait for all nodes to be healthy (auto-called by start)")]
wait timeout="180":
@@ -217,8 +292,8 @@ wait timeout="180":
NODES=$(just _node-list)
[ -z "$NODES" ] && sleep 5 && ELAPSED=$((ELAPSED + 5)) && continue
ALL_HEALTHY=true
- for n in $NODES; do
- S=$(docker inspect --format='{{ '{{' }}.State.Health.Status{{ '}}' }}' "$n" 2>/dev/null)
+ for svc in $NODES; do
+ S=$({{COMPOSE}} ps --format '{{ '{{' }}.Health{{ '}}' }}' "$svc" 2>/dev/null)
[ "$S" != "healthy" ] && ALL_HEALTHY=false && break
done
[ "$ALL_HEALTHY" = true ] && echo "✓ All nodes healthy" && exit 0
@@ -244,7 +319,7 @@ bin *ARGS:
*) PARAMS+=("$1"); shift ;;
esac
done
- docker exec "node-${NODE}" \
+ {{COMPOSE}} exec "node-${NODE}" \
env "KAFKA_HEAP_OPTS={{HEAP_OPTS}}" KAFKA_JVM_PERFORMANCE_OPTS="" JMX_PORT="" \
/opt/automq/bin/"${PARAMS[@]}" 2> >(grep -v '^SLF4J' >&2)
@@ -259,7 +334,7 @@ jmx *ARGS:
*) break ;;
esac
done
- echo "$@" | docker exec -i node-${NODE} java -jar /usr/local/bin/jmxterm.jar -l localhost:9999 -n
+ echo "$@" | {{COMPOSE}} exec -T node-${NODE} java -jar /usr/local/bin/jmxterm.jar -l localhost:9999 -n
# ==================== Shortcuts ====================
@@ -286,7 +361,7 @@ produce topic *ARGS:
*) PARAMS+=("$1"); shift ;;
esac
done
- docker exec -i "node-${NODE}" \
+ {{COMPOSE}} exec -T "node-${NODE}" \
env "KAFKA_HEAP_OPTS={{HEAP_OPTS}}" KAFKA_JVM_PERFORMANCE_OPTS="" JMX_PORT="" \
/opt/automq/bin/kafka-console-producer.sh \
--bootstrap-server localhost:9092 --topic "{{topic}}" \
@@ -303,7 +378,7 @@ consume topic *ARGS:
*) PARAMS+=("$1"); shift ;;
esac
done
- docker exec -i "node-${NODE}" \
+ {{COMPOSE}} exec -T "node-${NODE}" \
env "KAFKA_HEAP_OPTS={{HEAP_OPTS}}" KAFKA_JVM_PERFORMANCE_OPTS="" JMX_PORT="" \
/opt/automq/bin/kafka-console-consumer.sh \
--bootstrap-server localhost:9092 --topic "{{topic}}" \
@@ -352,24 +427,24 @@ perf-consume topic *ARGS:
[doc("Attach Arthas to Kafka process. Example: just arthas 1")]
arthas node="0":
- @docker exec -it node-{{node}} java -jar /opt/arthas/arthas-boot.jar
+ @{{COMPOSE}} exec node-{{node}} java -jar /opt/arthas/arthas-boot.jar
[doc("Run a single Arthas command non-interactively. Example: just arthas-exec 0 'dashboard -n 1'")]
arthas-exec node cmd:
#!/usr/bin/env bash
- printf '%s\n' '{{cmd}}' | docker exec -i node-{{node}} bash -c \
+ printf '%s\n' '{{cmd}}' | {{COMPOSE}} exec -T node-{{node}} bash -c \
'cat > /tmp/_arthas_cmd && java -jar /opt/arthas/arthas-boot.jar --select kafka.Kafka -f /tmp/_arthas_cmd'
# ==================== Chaos: Node network ====================
[doc("Network delay on node. Example: just chaos-delay 500 node-1")]
chaos-delay ms="200" node="node-0":
- docker exec {{node}} tc qdisc replace dev eth0 root netem delay {{ms}}ms
+ {{COMPOSE}} exec {{node}} tc qdisc replace dev eth0 root netem delay {{ms}}ms
@echo "✓ {{ms}}ms delay on {{node}}"
[doc("Packet loss on node. Example: just chaos-loss 10 node-1")]
chaos-loss percent="5" node="node-0":
- docker exec {{node}} tc qdisc replace dev eth0 root netem loss {{percent}}%
+ {{COMPOSE}} exec {{node}} tc qdisc replace dev eth0 root netem loss {{percent}}%
@echo "✓ {{percent}}% packet loss on {{node}}"
# ==================== Chaos: S3 (MinIO) ====================
@@ -377,45 +452,45 @@ chaos-loss percent="5" node="node-0":
[doc("S3 latency on a node (only S3 traffic affected). Example: just chaos-s3-delay 500 node-0")]
chaos-s3-delay ms="500" node="node-0":
#!/usr/bin/env bash
- MINIO_IP=$(docker exec {{node}} getent hosts minio | awk '{print $1}')
- docker exec {{node}} tc qdisc del dev eth0 root 2>/dev/null || true
- docker exec {{node}} tc qdisc add dev eth0 root handle 1: prio
- docker exec {{node}} tc qdisc add dev eth0 parent 1:3 handle 30: netem delay {{ms}}ms
- docker exec {{node}} tc filter add dev eth0 parent 1:0 protocol ip u32 match ip dst $MINIO_IP/32 flowid 1:3
+ MINIO_IP=$({{COMPOSE}} exec {{node}} getent hosts minio | awk '{print $1}')
+ {{COMPOSE}} exec {{node}} tc qdisc del dev eth0 root 2>/dev/null || true
+ {{COMPOSE}} exec {{node}} tc qdisc add dev eth0 root handle 1: prio
+ {{COMPOSE}} exec {{node}} tc qdisc add dev eth0 parent 1:3 handle 30: netem delay {{ms}}ms
+ {{COMPOSE}} exec {{node}} tc filter add dev eth0 parent 1:0 protocol ip u32 match ip dst $MINIO_IP/32 flowid 1:3
echo "✓ {{ms}}ms delay to MinIO on {{node}} (other traffic unaffected)"
[doc("S3 packet loss on a node (only S3 traffic affected). Example: just chaos-s3-loss 10 node-0")]
chaos-s3-loss percent="5" node="node-0":
#!/usr/bin/env bash
- MINIO_IP=$(docker exec {{node}} getent hosts minio | awk '{print $1}')
- docker exec {{node}} tc qdisc del dev eth0 root 2>/dev/null || true
- docker exec {{node}} tc qdisc add dev eth0 root handle 1: prio
- docker exec {{node}} tc qdisc add dev eth0 parent 1:3 handle 30: netem loss {{percent}}%
- docker exec {{node}} tc filter add dev eth0 parent 1:0 protocol ip u32 match ip dst $MINIO_IP/32 flowid 1:3
+ MINIO_IP=$({{COMPOSE}} exec {{node}} getent hosts minio | awk '{print $1}')
+ {{COMPOSE}} exec {{node}} tc qdisc del dev eth0 root 2>/dev/null || true
+ {{COMPOSE}} exec {{node}} tc qdisc add dev eth0 root handle 1: prio
+ {{COMPOSE}} exec {{node}} tc qdisc add dev eth0 parent 1:3 handle 30: netem loss {{percent}}%
+ {{COMPOSE}} exec {{node}} tc filter add dev eth0 parent 1:0 protocol ip u32 match ip dst $MINIO_IP/32 flowid 1:3
echo "✓ {{percent}}% packet loss to MinIO on {{node}} (other traffic unaffected)"
[doc("Pause MinIO — S3 completely unavailable")]
chaos-s3-down:
- docker pause minio
+ {{COMPOSE}} pause minio
@echo "✓ MinIO (S3) paused — all nodes lost S3 access"
[doc("Resume MinIO")]
chaos-s3-up:
- docker unpause minio
+ {{COMPOSE}} unpause minio
@echo "✓ MinIO (S3) resumed"
[doc("Isolate a node from S3. Example: just chaos-s3-partition node-1")]
chaos-s3-partition node="node-0": (_ensure-iptables-chain node)
- docker exec {{node}} iptables -A DEVKIT -d minio -j DROP
- docker exec {{node}} iptables -A DEVKIT -s minio -j DROP
+ {{COMPOSE}} exec {{node}} iptables -A DEVKIT -d minio -j DROP
+ {{COMPOSE}} exec {{node}} iptables -A DEVKIT -s minio -j DROP
@echo "✓ {{node}} isolated from MinIO (S3)"
[doc("Restore all nodes' S3 connectivity")]
chaos-s3-partition-reset:
#!/usr/bin/env bash
for n in $(just _node-list); do
- docker exec $n iptables -D DEVKIT -d minio -j DROP 2>/dev/null || true
- docker exec $n iptables -D DEVKIT -s minio -j DROP 2>/dev/null || true
+ {{COMPOSE}} exec $n iptables -D DEVKIT -d minio -j DROP 2>/dev/null || true
+ {{COMPOSE}} exec $n iptables -D DEVKIT -s minio -j DROP 2>/dev/null || true
done
echo "✓ All S3 partitions removed"
@@ -423,17 +498,17 @@ chaos-s3-partition-reset:
[doc("Isolate two nodes. Example: just chaos-partition node-0 node-1")]
chaos-partition a b: (_ensure-iptables-chain a) (_ensure-iptables-chain b)
- docker exec {{a}} iptables -A DEVKIT -d {{b}} -j DROP
- docker exec {{a}} iptables -A DEVKIT -s {{b}} -j DROP
- docker exec {{b}} iptables -A DEVKIT -d {{a}} -j DROP
- docker exec {{b}} iptables -A DEVKIT -s {{a}} -j DROP
+ {{COMPOSE}} exec {{a}} iptables -A DEVKIT -d {{b}} -j DROP
+ {{COMPOSE}} exec {{a}} iptables -A DEVKIT -s {{b}} -j DROP
+ {{COMPOSE}} exec {{b}} iptables -A DEVKIT -d {{a}} -j DROP
+ {{COMPOSE}} exec {{b}} iptables -A DEVKIT -s {{a}} -j DROP
@echo "✓ {{a}} ↔ {{b}} isolated"
[doc("Restore all node-to-node connectivity")]
chaos-partition-reset:
#!/usr/bin/env bash
for n in $(just _node-list); do
- docker exec $n iptables -F DEVKIT 2>/dev/null || true
+ {{COMPOSE}} exec $n iptables -F DEVKIT 2>/dev/null || true
done
echo "✓ All node partitions removed"
@@ -443,11 +518,11 @@ chaos-partition-reset:
chaos-reset:
#!/usr/bin/env bash
for n in $(just _node-list); do
- docker exec $n tc qdisc del dev eth0 root 2>/dev/null || true
- docker exec $n iptables -F DEVKIT 2>/dev/null || true
+ {{COMPOSE}} exec $n tc qdisc del dev eth0 root 2>/dev/null || true
+ {{COMPOSE}} exec $n iptables -F DEVKIT 2>/dev/null || true
done
# Unpause MinIO if paused
- docker unpause minio 2>/dev/null || true
+ {{COMPOSE}} unpause minio 2>/dev/null || true
echo "✓ All chaos rules reset (tc + iptables + unpause)"
[doc("Show all active chaos rules")]
@@ -455,10 +530,14 @@ chaos-status:
#!/usr/bin/env bash
for n in $(just _node-list); do
echo "=== $n ==="
- echo " tc:" && docker exec $n tc qdisc show dev eth0
- RULES=$(docker exec $n iptables -L DEVKIT -n 2>/dev/null | grep -c DROP || true)
+ echo " tc:" && {{COMPOSE}} exec $n tc qdisc show dev eth0
+ RULES=$({{COMPOSE}} exec $n iptables -L DEVKIT -n 2>/dev/null | grep -c DROP || true)
[ "$RULES" -gt 0 ] && echo " iptables: ${RULES} DROP rules in DEVKIT chain"
done
echo "=== minio ==="
- S=$(docker inspect --format='{{ '{{' }}.State.Paused{{ '}}' }}' minio 2>/dev/null)
- [ "$S" = "true" ] && echo " status: PAUSED" || echo " status: running"
+ S=$(docker inspect --format='{{ '{{' }}.State.Paused{{ '}}' }}' "$({{COMPOSE}} ps -q minio 2>/dev/null)" 2>/dev/null)
+ case "$S" in
+ true) echo " status: PAUSED" ;;
+ false) echo " status: running" ;;
+ *) echo " status: not running" ;;
+ esac
diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle
index 73990dbb34..79dd80fff6 100644
--- a/gradle/dependencies.gradle
+++ b/gradle/dependencies.gradle
@@ -98,7 +98,11 @@ versions += [
gradle: "8.10.2",
grgit: "4.1.1",
httpclient: "4.5.14",
- jackson: "2.17.1",
+ // AutoMQ inject start
+ jackson: "2.20.0",
+ // jackson-annotations uses "2.20" instead of "2.20.0"
+ jacksonAnnotationsVersion: "2.20",
+ // AutoMQ inject end
jacoco: "0.8.10",
javassist: "3.29.2-GA",
jetty: "9.4.57.v20241219",
@@ -146,7 +150,10 @@ versions += [
metrics: "2.2.0",
netty: "4.1.119.Final",
opentelemetryProto: "1.0.0-alpha",
- protobuf: "3.25.5", // a dependency of opentelemetryProto
+ // AutoMQ inject start
+ protobuf: "4.32.1",
+ protoGoogleCommonProtos: "2.40.0",
+ // AutoMQ inject end
pcollections: "4.0.1",
reflections: "0.10.2",
reload4j: "1.2.25",
@@ -160,6 +167,9 @@ versions += [
scoverage: "2.0.11",
slf4j: "1.7.36",
jclOverSlf4j: "1.7.36",
+ // AutoMQ inject start
+ snakeyaml: "2.5",
+ // AutoMQ inject end
snappy: "1.1.10.5",
spotbugs: "4.8.6",
zinc: "1.9.2",
@@ -172,18 +182,19 @@ versions += [
opentelemetrySDK: "1.40.0",
opentelemetrySDKAlpha: "1.40.0-alpha",
opentelemetryInstrument: "2.6.0-alpha",
- oshi: "6.4.7",
awsSdk:"2.29.26",
- bucket4j:"8.5.0",
+ bucket4j:"8.10.1",
jna:"5.2.0",
guava:"32.0.1-jre",
hdrHistogram:"2.1.12",
nettyTcnativeBoringSsl: "2.0.69.Final",
+ grpc: "1.65.0",
avro: "1.11.4",
confluentSchema: "7.8.0",
iceberg: "1.6.1",
wire: "4.9.1",
oshi: "6.8.1",
+ cloudevents: "4.0.1",
// AutoMQ inject end
// When updating the zstd version, please do as well in docker/native/native-image-configs/resource-config.json
@@ -209,7 +220,9 @@ libs += [
commonsCli: "commons-cli:commons-cli:$versions.commonsCli",
commonsIo: "commons-io:commons-io:$versions.commonsIo",
commonsValidator: "commons-validator:commons-validator:$versions.commonsValidator",
- jacksonAnnotations: "com.fasterxml.jackson.core:jackson-annotations:$versions.jackson",
+ // AutoMQ inject start
+ jacksonAnnotations: "com.fasterxml.jackson.core:jackson-annotations:$versions.jacksonAnnotationsVersion",
+ // AutoMQ inject end
jacksonDatabind: "com.fasterxml.jackson.core:jackson-databind:$versions.jackson",
jacksonDataformatCsv: "com.fasterxml.jackson.dataformat:jackson-dataformat-csv:$versions.jackson",
jacksonModuleScala: "com.fasterxml.jackson.module:jackson-module-scala_$versions.baseScala:$versions.jackson",
@@ -217,6 +230,10 @@ libs += [
jacksonAfterburner: "com.fasterxml.jackson.module:jackson-module-afterburner:$versions.jackson",
jacksonJaxrsJsonProvider: "com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:$versions.jackson",
jacksonYaml: "com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:$versions.jackson",
+ // AutoMQ inject start
+ snakeyaml: "org.yaml:snakeyaml:$versions.snakeyaml",
+ protoGoogleCommonProtos: "com.google.api.grpc:proto-google-common-protos:$versions.protoGoogleCommonProtos",
+ // AutoMQ inject end
jaxAnnotationApi: "javax.annotation:javax.annotation-api:$versions.jaxAnnotation",
jaxbApi: "javax.xml.bind:jaxb-api:$versions.jaxb",
jaxrsApi: "javax.ws.rs:javax.ws.rs-api:$versions.jaxrs",
@@ -310,8 +327,10 @@ libs += [
opentelemetryExporterLogging: "io.opentelemetry:opentelemetry-exporter-logging:$versions.opentelemetrySDK",
opentelemetryExporterProm: "io.opentelemetry:opentelemetry-exporter-prometheus:$versions.opentelemetrySDKAlpha",
opentelemetryExporterOTLP: "io.opentelemetry:opentelemetry-exporter-otlp:$versions.opentelemetrySDK",
+ opentelemetryExporterSenderJdk: "io.opentelemetry:opentelemetry-exporter-sender-jdk:$versions.opentelemetrySDK",
+ opentelemetryExporterSenderGrpcManagedChannel: "io.opentelemetry:opentelemetry-exporter-sender-grpc-managed-channel:$versions.opentelemetrySDK",
+ grpcNettyShaded: "io.grpc:grpc-netty-shaded:$versions.grpc",
opentelemetryJmx: "io.opentelemetry.instrumentation:opentelemetry-jmx-metrics:$versions.opentelemetryInstrument",
- oshi: "com.github.oshi:oshi-core-java11:$versions.oshi",
bucket4j: "com.bucket4j:bucket4j-core:$versions.bucket4j",
jna: "net.java.dev.jna:jna:$versions.jna",
guava: "com.google.guava:guava:$versions.guava",
@@ -319,4 +338,5 @@ libs += [
kafkaAvroSerializer: "io.confluent:kafka-avro-serializer:$versions.confluentSchema",
spotbugsAnnotations: "com.github.spotbugs:spotbugs-annotations:$versions.spotbugs",
oshi: "com.github.oshi:oshi-core:$versions.oshi",
+ cloudeventsKafka: "io.cloudevents:cloudevents-kafka:$versions.cloudevents",
]
diff --git a/gradle/spotbugs-exclude.xml b/gradle/spotbugs-exclude.xml
index 4e212c88e2..fe85573500 100644
--- a/gradle/spotbugs-exclude.xml
+++ b/gradle/spotbugs-exclude.xml
@@ -636,6 +636,10 @@ For a detailed description of spotbugs bug categories, see https://spotbugs.read
+
diff --git a/tools/src/main/java/org/apache/kafka/tools/ClusterEventsCommand.java b/tools/src/main/java/org/apache/kafka/tools/ClusterEventsCommand.java
new file mode 100644
index 0000000000..db33b453ed
--- /dev/null
+++ b/tools/src/main/java/org/apache/kafka/tools/ClusterEventsCommand.java
@@ -0,0 +1,268 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.ClusterEventTypeRegistry;
+import org.apache.kafka.clients.admin.ClusterEventsReader;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.CommandDefaultOptions;
+import org.apache.kafka.server.util.CommandLineUtils;
+
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.time.OffsetDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import io.cloudevents.CloudEvent;
+import joptsimple.ArgumentAcceptingOptionSpec;
+
+/**
+ * CLI tool for querying the {@code __automq_cluster_events} internal topic.
+ *
+ * When {@code --until} is not specified, runs in tail mode: continuously polls
+ * for new events until Ctrl-C or {@code --max-events} is reached.
+ *
+ *
+ * kafka-cluster-events.sh --bootstrap-server localhost:9092 \
+ * --type com.automq.ops.rebalance.summary --since -24h
+ *
+ */
+public class ClusterEventsCommand {
+
+ private static final Duration POLL_TIMEOUT = Duration.ofSeconds(2);
+
+ public static void main(String[] args) {
+ Exit.exit(mainNoExit(args, System.out));
+ }
+
+ static int mainNoExit(String[] args, PrintStream out) {
+ ClusterEventsCommandOptions opts = new ClusterEventsCommandOptions(args);
+ try {
+ opts.checkArgs();
+ } catch (Exception e) {
+ out.println("Error: " + e.getMessage());
+ try {
+ opts.parser.printHelpOn(out);
+ } catch (IOException ioe) {
+ // ignore
+ }
+ return 1;
+ }
+
+ Properties props = buildAdminProps(opts, out);
+ if (props == null) return 1;
+
+ try (Admin admin = Admin.create(props);
+ ClusterEventsReader reader = admin.describeClusterEvents(parseSinceMs(opts))) {
+ return pollLoop(reader, out, opts);
+ } catch (Exception e) {
+ out.println("Error: " + e.getMessage());
+ return 1;
+ }
+ }
+
+ private static Long parseSinceMs(ClusterEventsCommandOptions opts) {
+ return opts.options.has(opts.sinceOpt)
+ ? parseTimeArg(opts.options.valueOf(opts.sinceOpt)) : null;
+ }
+
+ private static int pollLoop(ClusterEventsReader reader, PrintStream out,
+ ClusterEventsCommandOptions opts) {
+ EventFilter filter = EventFilter.from(opts);
+ AtomicBoolean running = new AtomicBoolean(true);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> running.set(false)));
+
+ int count = 0;
+ while (running.get() && count < filter.maxEvents) {
+ List events = reader.poll(POLL_TIMEOUT);
+ for (CloudEvent event : events) {
+ if (filter.isPastUntil(event)) return 0;
+ if (filter.matches(event)) {
+ out.println(formatEvent(event));
+ if (++count >= filter.maxEvents) return 0;
+ }
+ }
+ if (!filter.tailMode && reader.isPolledToEnd()) break;
+ }
+ return 0;
+ }
+
+ private static final class EventFilter {
+ final Long untilMs;
+ final int maxEvents;
+ final String eventType;
+ final Pattern sourceRegex;
+ final Pattern subjectRegex;
+ final boolean tailMode;
+
+ EventFilter(Long untilMs, int maxEvents, String eventType,
+ Pattern sourceRegex, Pattern subjectRegex) {
+ this.untilMs = untilMs;
+ this.maxEvents = maxEvents;
+ this.eventType = eventType;
+ this.sourceRegex = sourceRegex;
+ this.subjectRegex = subjectRegex;
+ this.tailMode = untilMs == null;
+ }
+
+ static EventFilter from(ClusterEventsCommandOptions opts) {
+ Long untilMs = opts.options.has(opts.untilOpt)
+ ? parseTimeArg(opts.options.valueOf(opts.untilOpt)) : null;
+ int maxEvents = opts.options.has(opts.maxEventsOpt)
+ ? opts.options.valueOf(opts.maxEventsOpt) : Integer.MAX_VALUE;
+ String eventType = opts.options.has(opts.typeOpt)
+ ? opts.options.valueOf(opts.typeOpt) : null;
+ Pattern sourceRegex = opts.options.has(opts.sourceOpt)
+ ? Pattern.compile(opts.options.valueOf(opts.sourceOpt)) : null;
+ Pattern subjectRegex = opts.options.has(opts.subjectOpt)
+ ? Pattern.compile(opts.options.valueOf(opts.subjectOpt)) : null;
+ return new EventFilter(untilMs, maxEvents, eventType, sourceRegex, subjectRegex);
+ }
+
+ boolean isPastUntil(CloudEvent event) {
+ if (untilMs == null) return false;
+ java.time.OffsetDateTime time = event.getTime();
+ return time != null && time.toInstant().toEpochMilli() > untilMs;
+ }
+
+ boolean matches(CloudEvent event) {
+ if (event == null) return false;
+ if (eventType != null && !eventType.equals(event.getType())) return false;
+ if (!matchesPattern(sourceRegex, event.getSource() != null ? event.getSource().toString() : null))
+ return false;
+ return matchesPattern(subjectRegex, event.getSubject());
+ }
+
+ private static boolean matchesPattern(Pattern pattern, String value) {
+ if (pattern == null) return true;
+ return value != null && pattern.matcher(value).find();
+ }
+ }
+
+ private static Properties buildAdminProps(ClusterEventsCommandOptions opts, PrintStream out) {
+ Properties props = new Properties();
+ if (opts.options.has(opts.commandConfigOpt)) {
+ try {
+ props.putAll(Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt)));
+ } catch (IOException e) {
+ out.println("Error loading command config: " + e.getMessage());
+ return null;
+ }
+ }
+ props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, opts.options.valueOf(opts.bootstrapServerOpt));
+ return props;
+ }
+
+ /**
+ * Parse a time argument. Accepts:
+ *
+ * ISO-8601 datetime: {@code 2026-04-07T10:00:00Z}
+ * Relative: {@code -1h}, {@code -24h}, {@code -7d}
+ *
+ */
+ static long parseTimeArg(String value) {
+ value = value.trim();
+ Pattern relative = Pattern.compile("^-(\\d+)([hHdD])$");
+ Matcher m = relative.matcher(value);
+ if (m.matches()) {
+ long amount = Long.parseLong(m.group(1));
+ String unit = m.group(2).toLowerCase(java.util.Locale.ROOT);
+ long millis = unit.equals("h") ? amount * 3600_000L : amount * 86400_000L;
+ return System.currentTimeMillis() - millis;
+ }
+ try {
+ return OffsetDateTime.parse(value, DateTimeFormatter.ISO_OFFSET_DATE_TIME)
+ .toInstant().toEpochMilli();
+ } catch (DateTimeParseException e) {
+ throw new IllegalArgumentException("Cannot parse time value '" + value
+ + "'. Use ISO-8601 (e.g. 2026-04-07T10:00:00Z) or relative (e.g. -1h, -7d).");
+ }
+ }
+
+ private static String formatEvent(CloudEvent event) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("type=").append(event.getType());
+ sb.append(" source=").append(event.getSource());
+ if (event.getSubject() != null)
+ sb.append(" subject=").append(event.getSubject());
+ if (event.getTime() != null)
+ sb.append(" time=").append(event.getTime());
+ ClusterEventTypeRegistry.decode(event).ifPresentOrElse(
+ decoded -> sb.append(" data=").append(decoded),
+ () -> {
+ io.cloudevents.CloudEventData data = event.getData();
+ if (data != null) {
+ byte[] bytes = data.toBytes();
+ if (bytes != null)
+ sb.append(" data=").append(new String(bytes, java.nio.charset.StandardCharsets.UTF_8));
+ }
+ }
+ );
+ return sb.toString();
+ }
+
+ static final class ClusterEventsCommandOptions extends CommandDefaultOptions {
+ final ArgumentAcceptingOptionSpec bootstrapServerOpt;
+ final ArgumentAcceptingOptionSpec commandConfigOpt;
+ final ArgumentAcceptingOptionSpec typeOpt;
+ final ArgumentAcceptingOptionSpec sinceOpt;
+ final ArgumentAcceptingOptionSpec untilOpt;
+ final ArgumentAcceptingOptionSpec sourceOpt;
+ final ArgumentAcceptingOptionSpec subjectOpt;
+ final ArgumentAcceptingOptionSpec maxEventsOpt;
+
+ ClusterEventsCommandOptions(String[] args) {
+ super(args);
+ bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED: The Kafka server to connect to.")
+ .withRequiredArg().describedAs("server").ofType(String.class);
+ commandConfigOpt = parser.accepts("command-config",
+ "Property file containing configs to be passed to Admin Client.")
+ .withRequiredArg().describedAs("config file").ofType(String.class);
+ typeOpt = parser.accepts("type", "Filter by CloudEvent type (e.g. com.automq.ops.rebalance.summary).")
+ .withRequiredArg().ofType(String.class);
+ sinceOpt = parser.accepts("since",
+ "Only show events at or after this time. ISO-8601 or relative (e.g. -1h, -7d).")
+ .withRequiredArg().ofType(String.class);
+ untilOpt = parser.accepts("until",
+ "Only show events at or before this time. ISO-8601 or relative. "
+ + "If not set, runs in tail mode (continuously polls until Ctrl-C).")
+ .withRequiredArg().ofType(String.class);
+ sourceOpt = parser.accepts("source", "Filter by source (regex).")
+ .withRequiredArg().ofType(String.class);
+ subjectOpt = parser.accepts("subject", "Filter by subject (regex).")
+ .withRequiredArg().ofType(String.class);
+ maxEventsOpt = parser.accepts("max-events", "Maximum number of events to return.")
+ .withRequiredArg().ofType(Integer.class);
+ options = parser.parse(args);
+ }
+
+ void checkArgs() {
+ CommandLineUtils.checkRequiredArgs(parser, options, bootstrapServerOpt);
+ }
+ }
+}