From 8a8ce2b1306e2916013d4ca2514eabe1218728ca Mon Sep 17 00:00:00 2001 From: Gezi-lzq Date: Thu, 9 Apr 2026 17:26:33 +0800 Subject: [PATCH 1/5] build: align opensource dependency versions (#3275) * build: align opensource dependency versions * build: skip legacy netty alias in alignment rule * build: revert guava baseline bump * build: remove duplicate oshi version and lib definitions The oshi key was defined twice in both versions and libs blocks, with the later definition silently overriding the earlier one. Remove the stale oshi-core-java11:6.4.7 entries (deprecated since oshi 6.5) and keep oshi-core:6.8.1 which was already the effective resolution. No behavioral change. * Revert "build: remove duplicate oshi version and lib definitions" This reverts commit b47e96467c01209952a6adfbfa8c239e7d0dc5b8. * build: remove shadowed oshi version and lib entries versions.oshi and libs.oshi were each defined twice in the Groovy map. The earlier entries (oshi-core-java11:6.4.7) were silently overridden by the later ones (oshi-core:6.8.1) since commit 1119acc. Remove the shadowed entries to eliminate ambiguity. No behavioral change. --- build.gradle | 25 ++++++++++++++++++++----- gradle/dependencies.gradle | 26 ++++++++++++++++++++------ 2 files changed, 40 insertions(+), 11 deletions(-) diff --git a/build.gradle b/build.gradle index 3368733737..7b9e9af1ff 100644 --- a/build.gradle +++ b/build.gradle @@ -157,6 +157,16 @@ allprojects { // that are unrelated to the project dependencies, we should not change them if (name != "zinc") { resolutionStrategy { + // AutoMQ inject start + eachDependency { details -> + if (details.requested.group == 'io.netty' && + details.requested.name != 'netty' && + !details.requested.name.startsWith('netty-tcnative')) { + details.useVersion versions.netty + details.because('Release packaging must align all Netty modules to a single version') + } + } + // AutoMQ inject end force( // be explicit about the javassist dependency version instead of relying on the transitive version libs.javassist, @@ -164,6 +174,9 @@ allprojects { libs.scalaLibrary, libs.scalaReflect, libs.jacksonAnnotations, + // AutoMQ inject start + libs.snakeyaml, + // AutoMQ inject end // be explicit about the Netty dependency version instead of relying on the version set by // ZooKeeper (potentially older and containing CVEs) libs.nettyHandler, @@ -993,7 +1006,7 @@ project(':core') { // table topic start implementation ("org.apache.avro:avro:${versions.avro}") implementation ("org.apache.avro:avro-protobuf:${versions.avro}") - implementation('com.google.protobuf:protobuf-java:3.25.5') + implementation libs.protobuf implementation ("org.apache.iceberg:iceberg-core:${versions.iceberg}") implementation ("org.apache.iceberg:iceberg-api:${versions.iceberg}") implementation ("org.apache.iceberg:iceberg-data:${versions.iceberg}") @@ -1003,7 +1016,7 @@ project(':core') { implementation ("org.apache.iceberg:iceberg-nessie:${versions.iceberg}") implementation ("software.amazon.awssdk:glue:${versions.awsSdk}") implementation ("software.amazon.awssdk:s3tables:${versions.awsSdk}") - implementation 'software.amazon.s3tables:s3-tables-catalog-for-iceberg:0.1.0' + implementation 'software.amazon.s3tables:s3-tables-catalog-for-iceberg:0.1.8' implementation ('org.apache.hadoop:hadoop-common:3.4.1') { exclude group: 'org.eclipse.jetty', module: '*' @@ -1051,7 +1064,7 @@ project(':core') { // Wire Runtime for schema handling implementation ("com.squareup.wire:wire-schema:${versions.wire}") implementation ("com.squareup.wire:wire-runtime:${versions.wire}") - implementation 'com.google.api.grpc:proto-google-common-protos:2.52.0' + implementation libs.protoGoogleCommonProtos // > Protobuf ext end // table topic end @@ -2404,7 +2417,7 @@ project(':automq-metrics') { implementation('io.opentelemetry:opentelemetry-sdk:1.40.0') implementation("io.opentelemetry.semconv:opentelemetry-semconv:1.25.0-alpha") implementation("io.opentelemetry.instrumentation:opentelemetry-runtime-telemetry-java8:2.6.0-alpha") - implementation('com.google.protobuf:protobuf-java:3.25.5') + implementation libs.protobuf implementation('org.xerial.snappy:snappy-java:1.1.10.5') } @@ -2474,7 +2487,9 @@ project(':tools') { exclude group: 'org.apache.kafka', module: 'kafka-clients' } implementation libs.bucket4j - implementation libs.oshi + implementation (libs.oshi) { + exclude group: 'org.slf4j', module: 'slf4j-api' + } // AutoMQ inject end // for SASL/OAUTHBEARER JWT validation diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle index 73990dbb34..ce9b06fb14 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,9 +182,8 @@ 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", @@ -209,7 +218,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 +228,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", @@ -311,7 +326,6 @@ libs += [ opentelemetryExporterProm: "io.opentelemetry:opentelemetry-exporter-prometheus:$versions.opentelemetrySDKAlpha", opentelemetryExporterOTLP: "io.opentelemetry:opentelemetry-exporter-otlp:$versions.opentelemetrySDK", 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", From 8feeabdd7dcfcedb8cfb9bb42b35de49590311ab Mon Sep 17 00:00:00 2001 From: Gezi-lzq Date: Thu, 9 Apr 2026 17:30:08 +0800 Subject: [PATCH 2/5] feat(devkit): add namespace support for multi-instance isolation (#3295) (#3298) * feat(devkit): add namespace support for multi-instance isolation * docs(devkit): add namespace section to README * refactor(devkit): simplify namespace implementation - Remove container_name from docker-compose.yml (COMPOSE_PROJECT_NAME handles isolation) - Remove DEVKIT_PREFIX variable from set-ns and compose.env - Replace repetitive port variable assignments with PORT_DEFS array loop - Simplify PORT_OFFSET reading with grep instead of source - Simplify chaos-status minio check with case statement - Fix _node-list exit code when no containers running * feat(devkit): add python3 and uv to container image --- devkit/Dockerfile | 4 +- devkit/README.md | 27 ++++++ devkit/docker-compose.yml | 52 +++++------ devkit/justfile | 185 +++++++++++++++++++++++++++----------- 4 files changed, 183 insertions(+), 85 deletions(-) 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 From a1eda2eecefd01b0639ac1431a6a909787ac9273 Mon Sep 17 00:00:00 2001 From: Gezi-lzq Date: Mon, 13 Apr 2026 12:04:58 +0800 Subject: [PATCH 3/5] build: avoid forcing SnakeYAML for swagger doc generation (#3308) --- build.gradle | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/build.gradle b/build.gradle index 7b9e9af1ff..7e030bc116 100644 --- a/build.gradle +++ b/build.gradle @@ -174,9 +174,6 @@ allprojects { libs.scalaLibrary, libs.scalaReflect, libs.jacksonAnnotations, - // AutoMQ inject start - libs.snakeyaml, - // AutoMQ inject end // be explicit about the Netty dependency version instead of relying on the version set by // ZooKeeper (potentially older and containing CVEs) libs.nettyHandler, @@ -184,6 +181,12 @@ allprojects { // be explicit about the reload4j version instead of relying on the transitive versions libs.reload4j ) + // AutoMQ inject start + // swagger is only used for doc generation and still depends on jackson-yaml compatible with SnakeYAML 1.x. + if (name != "swagger") { + force(libs.snakeyaml) + } + // AutoMQ inject end } } } From 656e77936921e6584ee369fae51989ecdcc2cc52 Mon Sep 17 00:00:00 2001 From: Gezi-lzq Date: Mon, 13 Apr 2026 15:25:43 +0800 Subject: [PATCH 4/5] fix: replace sender-okhttp with sender-grpc-managed-channel for Strimzi compatibility (#3268) (#3309) cherry-pick: #3268 d63afdf Co-authored-by: rafaelsoutto --- build.gradle | 8 +++++++- gradle/dependencies.gradle | 4 ++++ 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/build.gradle b/build.gradle index 7e030bc116..891adc917e 100644 --- a/build.gradle +++ b/build.gradle @@ -179,7 +179,12 @@ allprojects { libs.nettyHandler, libs.nettyTransportNativeEpoll, // be explicit about the reload4j version instead of relying on the transitive versions - libs.reload4j + libs.reload4j, + // AutoMQ inject start + // pin OTel sender versions to prevent version conflicts with Strimzi thirdparty-libs + libs.opentelemetryExporterSenderJdk, + libs.opentelemetryExporterSenderGrpcManagedChannel + // AutoMQ inject end ) // AutoMQ inject start // swagger is only used for doc generation and still depends on jackson-yaml compatible with SnakeYAML 1.x. @@ -2379,6 +2384,7 @@ project(':automq-metrics') { api libs.opentelemetryExporterLogging api libs.opentelemetryExporterProm api libs.opentelemetryExporterOTLP + api libs.opentelemetryExporterSenderJdk api libs.opentelemetryJmx // Logging dependencies diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle index ce9b06fb14..69f6e59230 100644 --- a/gradle/dependencies.gradle +++ b/gradle/dependencies.gradle @@ -188,6 +188,7 @@ versions += [ 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", @@ -325,6 +326,9 @@ 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", bucket4j: "com.bucket4j:bucket4j-core:$versions.bucket4j", jna: "net.java.dev.jna:jna:$versions.jna", From 8d77482f22e6a96d4398a06c5c85f5ad02170092 Mon Sep 17 00:00:00 2001 From: Robin Han Date: Tue, 14 Apr 2026 14:21:36 +0800 Subject: [PATCH 5/5] feat: add cluster events framework with protobuf-based event publishing Introduce a cluster-level event system that publishes structured events to the __automq_cluster_events internal topic using CloudEvents + protobuf. Core infrastructure: - ClusterEventPublisher: async, best-effort producer that wraps events as CloudEvents and writes them to the internal topic - ClusterEventsReader: consumer-based reader for retrieving events - ClusterEventsCommand: CLI tool (kafka-cluster-events.sh) for reading and displaying cluster events RequestErrorEvent pipeline: - Detects sustained request error patterns (e.g. auth failures, authorization errors) on the broker and publishes them as events, enabling the control plane to surface risky or misconfigured clients - Integrated into RequestChannel and KafkaApis to capture errors on the broker request path - ResourceErrorExtractor parses per-resource error codes from Kafka API responses (Produce, Fetch, Metadata, etc.) - RequestErrorAccumulator aggregates errors in-memory by (apiKey, errorCode, resource) with client IP/ID tracking, then periodically flushes accumulated buckets as RequestErrorEvent records --- .gitignore | 4 + bin/kafka-cluster-events.sh | 17 + build.gradle | 34 +- .../org/apache/kafka/clients/admin/Admin.java | 8 + .../clients/admin/ClusterEventPublisher.java | 173 ++++++++ .../admin/ClusterEventTypeRegistry.java | 78 ++++ .../clients/admin/ClusterEventsConfig.java | 29 ++ .../clients/admin/ClusterEventsReader.java | 133 ++++++ .../clients/admin/FailoverEventData.java | 102 +++++ .../kafka/clients/admin/ForwardingAdmin.java | 5 + .../clients/admin/IClusterEventPublisher.java | 38 ++ .../kafka/clients/admin/KafkaAdminClient.java | 16 +- .../admin/NoopClusterEventPublisher.java | 37 ++ .../admin/OffsetCommitFrequencyEventData.java | 119 +++++ .../admin/RebalancePartitionEventData.java | 115 +++++ .../admin/RebalanceSummaryEventData.java | 118 +++++ .../clients/admin/RequestErrorEventData.java | 159 +++++++ .../apache/kafka/common/internals/Topic.java | 1 + .../com/automq/events/cluster_events.proto | 70 +++ .../admin/ClusterEventTypeRegistryTest.java | 164 +++++++ .../kafka/clients/admin/MockAdminClient.java | 5 + .../kafka/server/RequestErrorAccumulator.java | 135 ++++++ .../kafka/server/ResourceErrorExtractor.java | 408 ++++++++++++++++++ .../scala/kafka/network/RequestChannel.scala | 52 ++- .../server/AutoTopicCreationManager.scala | 12 +- .../scala/kafka/server/BrokerServer.scala | 23 + .../main/scala/kafka/server/KafkaApis.scala | 6 +- .../server/RequestErrorEventDataTest.java | 51 +++ gradle/dependencies.gradle | 2 + gradle/spotbugs-exclude.xml | 4 + .../kafka/tools/ClusterEventsCommand.java | 268 ++++++++++++ 31 files changed, 2377 insertions(+), 9 deletions(-) create mode 100755 bin/kafka-cluster-events.sh create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventPublisher.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventTypeRegistry.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventsConfig.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventsReader.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/FailoverEventData.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/IClusterEventPublisher.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/NoopClusterEventPublisher.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/OffsetCommitFrequencyEventData.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/RebalancePartitionEventData.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/RebalanceSummaryEventData.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/RequestErrorEventData.java create mode 100644 clients/src/main/proto/com/automq/events/cluster_events.proto create mode 100644 clients/src/test/java/org/apache/kafka/clients/admin/ClusterEventTypeRegistryTest.java create mode 100644 core/src/main/java/kafka/server/RequestErrorAccumulator.java create mode 100644 core/src/main/java/kafka/server/ResourceErrorExtractor.java create mode 100644 core/src/test/java/kafka/server/RequestErrorEventDataTest.java create mode 100644 tools/src/main/java/org/apache/kafka/tools/ClusterEventsCommand.java diff --git a/.gitignore b/.gitignore index 42f611bb67..5dc6f08a94 100644 --- a/.gitignore +++ b/.gitignore @@ -67,3 +67,7 @@ __pycache__ bin/ !/bin/ release/.venv/ + +# Claude Code project config +.claude/settings.local.json +.claude/plans/ diff --git a/bin/kafka-cluster-events.sh b/bin/kafka-cluster-events.sh new file mode 100755 index 0000000000..b9ce15ca13 --- /dev/null +++ b/bin/kafka-cluster-events.sh @@ -0,0 +1,17 @@ +#!/bin/bash +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.ClusterEventsCommand "$@" diff --git a/build.gradle b/build.gradle index 891adc917e..b9622f9ae2 100644 --- a/build.gradle +++ b/build.gradle @@ -44,6 +44,7 @@ plugins { // Updating the shadow plugin version to 8.1.1 causes issue with signing and publishing the shadowed // artifacts - see https://github.com/johnrengelman/shadow/issues/901 id 'com.github.johnrengelman.shadow' version '8.1.0' apply false + id 'com.google.protobuf' version '0.9.4' apply false // Spotless 6.13.0 has issue with Java 21 (see https://github.com/diffplug/spotless/pull/1920), and Spotless 6.14.0+ requires JRE 11 // We are going to drop JDK8 support. Hence, the spotless is upgrade to newest version and be applied only if the build env is compatible with JDK 11. // spotless 6.15.0+ has issue in runtime with JDK8 even through we define it with `apply:false`. see https://github.com/diffplug/spotless/issues/2156 for more details @@ -789,7 +790,7 @@ subprojects { apply plugin: 'com.diffplug.spotless' spotless { java { - targetExclude('src/generated/**/*.java','src/generated-test/**/*.java') + targetExclude('src/generated/**/*.java', 'src/generated-test/**/*.java', 'build/generated/**/*.java') importOrder('kafka', 'org.apache.kafka', 'com', 'net', 'org', 'java', 'javax', '', '\\#') removeUnusedImports() } @@ -1704,6 +1705,8 @@ project(':generator') { } project(':clients') { + apply plugin: 'com.google.protobuf' + base { archivesName = "kafka-clients" } @@ -1721,6 +1724,10 @@ project(':clients') { implementation libs.opentelemetryProto implementation libs.protobuf + // AutoMQ inject start + implementation libs.cloudeventsKafka + // AutoMQ inject end + // libraries which should be added as runtime dependencies in generated pom.xml should be defined here: shadowed libs.zstd shadowed libs.lz4 @@ -1839,7 +1846,7 @@ project(':clients') { sourceSets { main { java { - srcDirs = ["src/generated/java", "src/main/java"] + srcDirs = ["src/generated/java", "src/main/java", "$buildDir/generated/source/proto/main/java"] // AutoMQ: add proto source } } test { @@ -1849,7 +1856,28 @@ project(':clients') { } } + protobuf { + protoc { + artifact = "com.google.protobuf:protoc:$versions.protobuf" + } + } + + // Exclude protobuf-generated sources from checkstyle and spotless + afterEvaluate { + checkstyleMain.source = checkstyleMain.source.filter { !it.path.contains('/generated/source/proto/') } + if (tasks.findByName('spotlessJava')) { + spotless { + java { + targetExclude('build/generated/source/proto/**/*.java') + } + } + } + } + compileJava.dependsOn 'processMessages' + // AutoMQ inject start: ensure proto sources are generated before compilation + compileJava.dependsOn 'generateProto' + // AutoMQ inject end srcJar.dependsOn 'processMessages' compileTestJava.dependsOn 'processTestMessages' @@ -2499,6 +2527,8 @@ project(':tools') { implementation (libs.oshi) { exclude group: 'org.slf4j', module: 'slf4j-api' } + implementation libs.cloudeventsKafka + implementation libs.protobuf // AutoMQ inject end // for SASL/OAUTHBEARER JWT validation diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java index b634366d1e..f93e97bffc 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java @@ -1757,6 +1757,14 @@ default ExportClusterManifestResult exportClusterManifest() { } ExportClusterManifestResult exportClusterManifest(ExportClusterManifestOptions options); + + /** + * Create a reader for the {@code __automq_cluster_events} internal topic. + * + * @param sinceMs only return events at or after this epoch-millisecond timestamp, or null for all + * @return a {@link ClusterEventsReader} that must be closed when done + */ + ClusterEventsReader describeClusterEvents(Long sinceMs); // AutoMQ inject end /** diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventPublisher.java b/clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventPublisher.java new file mode 100644 index 0000000000..53de04ea05 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/admin/ClusterEventPublisher.java @@ -0,0 +1,173 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.clients.admin; + +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.internals.Topic; +import org.apache.kafka.common.serialization.StringSerializer; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.net.URI; +import java.time.OffsetDateTime; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import io.cloudevents.CloudEvent; +import io.cloudevents.core.builder.CloudEventBuilder; +import io.cloudevents.kafka.CloudEventSerializer; + +/** + * Publishes cluster events to the {@code __automq_cluster_events} internal topic. + * + *

Events are written asynchronously and best-effort: if the producer fails to send an event, + * a warning is logged and the event is dropped. Events must never block or fail the critical path + * (rebalancing, failover, request handling). + * + *

Example usage: + * 

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

Usage: + * 

{@code
+ * ClusterEventTypeRegistry.decode(event).ifPresent(obj -> {
+ *     if (obj instanceof RebalanceSummaryEventData summary) { ... }
+ *     else if (obj instanceof FailoverEventData failover) { ... }
+ * });
+ * }
+ */ +public class ClusterEventTypeRegistry { + + private static final Logger log = LoggerFactory.getLogger(ClusterEventTypeRegistry.class); + + private static final Map> DECODERS = Map.of( + RebalanceSummaryEventData.DATA_SCHEMA, b -> RebalanceSummaryEventData.fromByteArray(b), + RebalancePartitionEventData.DATA_SCHEMA, b -> RebalancePartitionEventData.fromByteArray(b), + FailoverEventData.DATA_SCHEMA, b -> FailoverEventData.fromByteArray(b), + RequestErrorEventData.DATA_SCHEMA, b -> RequestErrorEventData.fromByteArray(b), + OffsetCommitFrequencyEventData.DATA_SCHEMA, b -> OffsetCommitFrequencyEventData.fromByteArray(b) + ); + + private ClusterEventTypeRegistry() { } + + /** + * Decode the protobuf payload of a CloudEvent into its POJO wrapper. + * + * @return the decoded wrapper, or empty if the dataschema is unknown or the payload is invalid + */ + public static Optional decode(CloudEvent event) { + URI dataSchema = event.getDataSchema(); + CloudEventData data = event.getData(); + if (dataSchema == null || data == null) { + return Optional.empty(); + } + Function 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/gradle/dependencies.gradle b/gradle/dependencies.gradle index 69f6e59230..79dd80fff6 100644 --- a/gradle/dependencies.gradle +++ b/gradle/dependencies.gradle @@ -194,6 +194,7 @@ versions += [ 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 @@ -337,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); + } + } +}