Dynamic metric#3455
Open
sachin428z wants to merge 112 commits into
Open
Conversation
fix: add missing jdk http sender for otlp metrics exporter Signed-off-by: Adhiraj <68840640+adhraj12@users.noreply.github.com>
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
AutoMQ#3150) fix(metrics): enhance updatePartitionBytesIn to support custom count parameter
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
Added a quick start section for first-time contributors, outlining two onboarding paths: - quick exploration using Docker - local development setup for code contributors Signed-off-by: Shreyas Ranjan <shreyasranjan4676@gmail.com>
…Q#3156) fix(connect): do not print stack when s3 error
…toMQ#3160) Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
fix(e2e): change runner
…ts (AutoMQ#3166) * feat(router): implement RouterPermitLimiter for managing append permits * fix(router): update permit handling in RouterPermitLimiter and related classes * fix(router): refactor permit handling to use environment variables for append permit size * fix(zerozone): optimize permit acquisition logic in RouterInV2 * fix(router): remove unused Systems import from RouterPermitLimiter
…ch (AutoMQ#3172) fix(perf): fix backlog benchmark exits immediately due to time base mismatch
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
* docs: Add GSoC 2026 documentation * Update AutoMQ_ideas_list.md Signed-off-by: cneym1125 <122273191+cneym1125@users.noreply.github.com> * Update Contributor_Guidance_for_Google_Summer_of_Code.md Signed-off-by: cneym1125 <122273191+cneym1125@users.noreply.github.com> --------- Signed-off-by: cneym1125 <122273191+cneym1125@users.noreply.github.com> Co-authored-by: cneym1125 <3215663031aaa@gmail.com>
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
feat(license): add license support (AutoMQ#3136)
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
* feat(devbox): add zero-config local development environment Replace local-playground with devbox — a lightweight, layered local dev environment for AutoMQ developers. Features: - just start / just start-build: single node or cluster (up to 5 nodes) - Docker Compose profiles: single, cluster, tabletopic, analytics - 4-layer config system: defaults → features → custom → auto-generated - just bin: direct passthrough to any bin/ command in container - Shortcuts: topic-list, topic-create, produce, consume - Chaos engineering: chaos-delay, chaos-loss, chaos-reset - JDWP remote debugging on every node - Stable auto-generated CLUSTER_ID Dependencies: just + docker + bash * fix(devbox): correct spacing in topic description output * feat(devbox): update CI to test 4-node cluster configuration * fix(devbox): update CI to depend on lint instead of smoke for full job * feat(devbox): add concurrency settings to CI workflow * fix(devbox): remove interactive flag from docker exec command * feat(devbox): update advertised listeners for controller nodes in configuration * docs(devbox): update README to clarify role and node identity configuration * refactor(devbox): improve security, CI coverage, and documentation - Remove privileged flag, use NET_ADMIN capability only - Add Docker image build to build command - Clean up apt lists in Dockerfile to reduce image size - Remove unused SKIP variable from bin command - Add tabletopic and analytics profiles to CI validation - Remove duplicate 4-node cluster test - Fix PartitionCount grep pattern (add space) - Update documentation to clarify build includes image and code - Add note about container-internal access design - Add concurrency control to cancel old CI runs * feat(devbox): refactor justfile to use KAFKA_OPTS for Kafka commands * feat(devbox): add JMX operations and update configurations for Kafka * feat(devbox): update KAFKA_OPTS to include JMX_PORT for improved monitoring * feat(devbox): set JMX_PORT for Kafka containers and enhance restart command * feat(devbox): set JMX_PORT in docker-compose for Kafka services * feat(devbox): add build-image command to rebuild Docker images with specified nodes
cherry-pick apache/kafka#17542 cherry-pick apache/kafka#20945 Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
This PR adds AutoMQ retry storm backoff for broker data-plane responses. It introduces a broker-side response gate that can delay all-error responses when a client is repeatedly receiving retryable/transient failures, or when the same request dimension is returning errors fast enough to become protective-load risk. The feature is disabled by default and can be enabled dynamically with `automq.retry.storm.backoff.enabled`. The strategy is intentionally conservative: - Scope: delay only selected broker responses after the response has already been built. The original Kafka error code, response schema, quota throttle semantics, and business semantics are preserved. - Decision input: `ResourceErrorExtractor` builds a reusable `ResourceErrorView` from the response, including per-resource key, error code, and whether the response is an all-error candidate. This is shared with existing request error accumulation instead of re-parsing for retry storm only. - Tracking dimension: `apiKey + resourceKey + connectionId`. This keeps different resources and client connections isolated. - Fast transient protection: delayable transient errors use a low threshold. The first failure is immediate; the second failure in the same dimension can be delayed. - Protective protection: non-transient or mixed all-error responses can enter a higher threshold protective path. The sixth error within the 1s window can be delayed. - Batch handling: partial success or any valid result returns immediately and does not update retry storm state. For all-error batches, each resource updates and decides independently, and the response-level delay uses the maximum delay from delayed resource decisions. - Recovery: state recovery is based on quiet timeout, not successful responses, so success-heavy paths avoid extra state mutation work. - State management: retry storm state is held in a bounded Guava cache with TTL and maximum tracked dimensions. Delaying-state logging scans cache entries without refreshing access time. - Observability: a periodic sampled logger prints at most 10 dimensions per minute that have been delaying for more than 10s, including API key, resource key, client scope, reason bitmask as strings, delayingSinceMs, and lastFailureMs. - Safety: the response gate fails open. If retry storm evaluation throws, the original response is sent immediately. Configuration: - `automq.retry.storm.backoff.enabled`: dynamic broker config, default `false`. - `automq.retry.storm.backoff.max.delay.ms`: dynamic broker config, default `1000`, valid range `[0, 10000]`. `0` means decisions may update state but responses are not scheduled for delay. Suggested review order: 1. Config and lifecycle - `core/src/main/java/kafka/automq/AutoMQConfig.java` - `core/src/main/java/kafka/automq/retrystorm/RetryStormBackoffConfig.java` - `core/src/main/java/kafka/automq/retrystorm/RetryStormBackoffManager.java` - `core/src/main/scala/kafka/server/BrokerServer.scala` - `core/src/main/scala/kafka/server/KafkaConfig.scala` - `core/src/main/scala/kafka/server/DynamicBrokerConfig.scala` 2. Response extraction and existing request error accumulator integration - `core/src/main/java/kafka/server/ResourceErrorExtractor.java` - `core/src/main/scala/kafka/network/RequestChannel.scala` 3. Backoff decision and state machine - `core/src/main/java/kafka/automq/retrystorm/RetryStormBackoffStateStore.java` - `core/src/main/java/kafka/server/retrystorm/RetryStormBackoffPolicy.java` - `core/src/main/java/kafka/server/retrystorm/BackoffDecision.java` - `core/src/main/java/kafka/server/retrystorm/BackoffContext.java` 4. Response scheduling/gating and fail-open behavior - `core/src/main/java/kafka/server/retrystorm/RetryStormResponseGate.java` - `core/src/main/java/kafka/server/retrystorm/RetryStormDelayedResponseScheduler.java` - `core/src/main/scala/kafka/server/RequestHandlerHelper.scala` - `core/src/main/scala/kafka/server/streamaspect/ElasticKafkaApis.scala` 5. Periodic sampled logging - `core/src/main/java/kafka/server/retrystorm/SampledRetryStormBackoffLogger.java` - `core/src/main/java/kafka/server/retrystorm/RetryStormBackoffLogger.java` 6. Tests and E2E - `core/src/test/java/kafka/server/ResourceErrorExtractorTest.java` - `core/src/test/java/kafka/server/retrystorm/RetryStormBackoffPolicyTest.java` - `core/src/test/java/kafka/automq/retrystorm/RetryStormBackoffStateStoreTest.java` - `core/src/test/java/kafka/network/RequestChannelRetryStormTest.java` - `tests/kafkatest/automq/retry_storm_backoff_test.py` - `tools/src/main/java/org/apache/kafka/tools/automq/RetryStormBackoffProbe.java` Reviewer focus areas: - Confirm all successful or partially successful responses remain immediate and do not update retry storm state. - Confirm mixed all-error batches use per-resource decisions and response-level max delay aggregation. - Confirm retry storm state is bounded and quiet-time recovery does not rely on success-path updates. - Confirm quota/throttle behavior and shutdown behavior remain compatible with existing Kafka response handling. - Confirm default-off compatibility is acceptable for the 1.7 branch. Fresh local checks before opening this PR: - `git diff --check origin/1.7...HEAD` passed. - `python3 -m py_compile tests/kafkatest/automq/retry_storm_backoff_test.py` passed. - `./gradlew tools:compileJava` passed. - `./gradlew core:S3UnitTest --tests kafka.automq.retrystorm.RetryStormBackoffConfigTest --tests kafka.automq.retrystorm.RetryStormBackoffStateStoreTest --tests kafka.automq.retrystorm.RetryStormBackoffManagerTest --tests kafka.server.DynamicBrokerConfigTest.testRetryStormBackoffConfigsAreDynamic --tests kafka.server.ResourceErrorExtractorTest --tests kafka.network.RequestChannelRetryStormTest --tests kafka.server.retrystorm.RetryStormBackoffPolicyTest --tests kafka.server.retrystorm.RetryStormDelayedResponseSchedulerTest --tests kafka.server.retrystorm.RetryStormResponseGateTest` passed. Previous broader validation recorded during the change: - `./gradlew --build-cache metadata:S3UnitTest core:S3UnitTest s3stream:test` passed after the main implementation and review-gate fixes. - Devkit validation passed with dynamic enable/disable and max-delay update behavior. - E2E test was added to `tests/suites/automq_test_suite1.yml`; full ducktape execution was prepared but not run locally in this session.
## Summary - declare `:clients:srcJar` dependency on `generateProto` - fix Gradle implicit dependency validation when publishing source jars with generated protobuf sources ## Verification - `./gradlew --build-cache :clients:srcJar -PskipSigning=true` - `./gradlew --build-cache rat checkstyleMain checkstyleTest spotlessJavaCheck` Related failure: https://github.com/AutoMQ/automq/actions/runs/26952811949/job/80015750009
…Q#3411) ## Summary - Add authorization checks for AutoMQ controller extension and broker-local API keys. - Map license/config-style APIs to `ALTER_CONFIGS` / `DESCRIBE_CONFIGS`, group update to group `READ`, and other AutoMQ extension APIs to cluster `CLUSTER_ACTION`. - Add focused authorization tests and fix extension error responses needed by denied paths.
## Summary Cherry-pick the static consumer fencing test fix from AutoMQ#3400 to `main`. This keeps `main` aligned with the Apache Kafka upstream fix in apache/kafka#20772 and the already-merged AutoMQ `1.7` backport. ## Root Cause `OffsetValidationTest.test_fencing_static_consumer` reuses the same static `group.instance.id` values for conflict consumers after stopping the original static members. For the consumer group protocol path, the conflict consumers are expected to fail with `UnreleasedInstanceIdException`. The old test flow did not wait for those conflict consumers to fully terminate before stopping the original consumers and restarting the conflict consumers with the same instance ids. That left a race where the old conflict processes could still hold or immediately retry the same static member identities. Apache Kafka fixed this in apache/kafka#20772 by waiting for the fenced conflict consumers and for the group to become empty before reusing the static instance ids. AutoMQ `1.7` already carries the same fix via AutoMQ#3400. `main` did not yet contain it. ## Changes - Track verifiable consumer shutdown completion explicitly. - Add a helper to wait for all members to reach a stable assignment. - In the consumer group protocol branch, wait for conflict consumers to be fenced and for the group to become empty before restarting them. - Extend `KafkaService.list_consumer_groups` with `--state` and `--type` filters needed by the test. - Mark the E2E-specific backport with `# // AutoMQ inject start` / `# // AutoMQ inject end`. ## Verification - `python3 -m py_compile tests/kafkatest/tests/client/consumer_test.py tests/kafkatest/tests/verifiable_consumer_test.py tests/kafkatest/services/kafka/kafka.py tests/kafkatest/services/verifiable_consumer.py` - `git diff --check HEAD~1..HEAD` The same fix was locally verified for the original failing Enterprise E2E matrix before merging AutoMQ#3400 into `1.7`.
## Summary This PR hardens S3 object writes against dirty buffer reads during SDK-level retries. ## Bug Flow AutoMQ writes S3 request bodies with `AsyncRequestBody.fromByteBuffersUnsafe(...)` to avoid copying Netty `ByteBuf` data. A rare corruption flow can happen when an SDK attempt times out but the underlying HTTP write is still alive: 1. AutoMQ submits a PutObject or UploadPart request backed by unsafe `ByteBuffer` views of a Netty `ByteBuf`. 2. The SDK attempt times out and starts a retry, while the first low-level request may still continue writing from the original buffers. 3. A later retry succeeds and completes the future. 4. The caller releases the `ByteBuf`; Netty may recycle the memory for another buffer. 5. The stale first request can still read from the recycled memory and send dirty bytes. 6. Without a stable precomputed checksum attached to the request, object storage may accept and persist the corrupted body. ## How This PR Fixes It The fix is to bind each write request to the bytes that were present before the unsafe buffers are passed to the SDK. For PutObject and UploadPart, AutoMQ now computes the request checksum synchronously while it still owns a valid `ByteBuf`. That checksum becomes the immutable expected value for the request body: - If the SDK later retries normally using the same original bytes, the object storage service receives bytes matching the precomputed checksum and accepts the write. - If an earlier timed-out HTTP attempt keeps running after the successful retry and reads from recycled Netty memory, it sends bytes that no longer match the precomputed checksum. The service should reject that stale request instead of persisting corrupted data. This is why the checksum must be computed before `AsyncRequestBody.fromByteBuffersUnsafe(...)` is handed to the SDK. Letting the SDK calculate a checksum from the unsafe body during the actual HTTP attempt would not fully protect this flow, because the checksum calculation could observe the same dirty/recycled memory as the stale request. The implementation uses the strongest available request-side checksum for the configured mode: - When a supported S3 flexible checksum algorithm is configured, AutoMQ precomputes and sets the concrete checksum header (`checksumCRC32`, `checksumCRC32C`, `checksumSHA1`, or `checksumSHA256`) on PutObject and UploadPart. - When no flexible checksum algorithm is configured, AutoMQ precomputes and sets `Content-MD5`. - For multipart uploads, CreateMultipartUpload still carries `checksumAlgorithm` to define the upload's checksum algorithm, and CompleteMultipartUpload sends the returned per-part checksum in the matching part checksum field instead of assuming CRC32C. - The SDK's legacy S3 ETag MD5 validation path is disabled to avoid a second client-side MD5 pass after AutoMQ already attaches its own request checksum.
…broker crash (AutoMQ#3418) ### Description When using AutoMQ with S3-compatible storage backends, a `NoSuchUploadException` can occur during an `UploadPart` or `UploadPartCopy` operation (e.g., if the multipart upload has expired, been aborted, or cleaned up by the backend). Previously, `AwsObjectStorage.toRetryStrategyAndCause` only handled `NoSuchUploadException` for `COMPLETE_MULTI_PART_UPLOAD` (mapping it to `RetryStrategy.VISIBILITY_CHECK`). For `UPLOAD_PART` and `UPLOAD_PART_COPY` operations, the exception fell through to the default `RetryStrategy.RETRY`. This caused an infinite retry loop, ultimately leading to a broker crash. This PR addresses the issue by mapping `NoSuchUploadException` to `RetryStrategy.ABORT` for both `UPLOAD_PART` and `UPLOAD_PART_COPY` operations. Since the multipart upload ID is no longer valid, aborting early prevents the crash loop. Fixes AutoMQ#3206 ### Testing Strategy - Added new unit tests in [AwsObjectStorageTest](file:///Users/divyanshuyadav/Downloads/automq/s3stream/src/test/java/com/automq/stream/s3/operator/AwsObjectStorageTest.java) to verify retry strategy mapping: - `testNoSuchUploadExceptionAbortsUploadPart` verifies `UPLOAD_PART` maps to `RetryStrategy.ABORT`. - `testNoSuchUploadExceptionAbortsUploadPartCopy` verifies `UPLOAD_PART_COPY` maps to `RetryStrategy.ABORT`. - `testNoSuchUploadExceptionVisibilityCheckForComplete` verifies `COMPLETE_MULTI_PART_UPLOAD` still maps to `RetryStrategy.VISIBILITY_CHECK`. - Ran the unit tests locally: ```bash ./gradlew :s3stream:test --tests "com.automq.stream.s3.operator.AwsObjectStorageTest" ### Committer Checklist (excluded from commit message) - [x] Verify design and implementation - [x] Verify test coverage and CI build status - [x] Verify documentation (including upgrade notes)
…Q#3425) - make CLEANUP_V1 bypass general grouping and only compact the first dirty composite object after expired cleanup - remove obsolete CLEANUP_V1 branches from shared compaction filter/group validation paths - trigger MAJOR_V1 at 90% of the object-count threshold while preserving small-normal-object skipping before the hard threshold
## Summary - Align shared Prometheus name normalization with Prometheus pull rules. - Keep `objects_search_count` pseudo-histogram names stable by removing the inappropriate `count` unit. - Keep the focused rule regression test in `automq-metrics`.
…utoMQ#3428) - release pooled records created during LocalLog offset metadata lookup - add regression coverage for releasing records returned by the internal read
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
## Summary Reduce storage pressure from E2E workflow artifacts. Changes: - Shorten E2E artifact retention from 3 days to 1 day. - Increase artifact compression level from 1 to 6. - Add a short comment explaining why E2E artifacts use a short retention window. ## Background The scheduled Nightly E2E workflow can upload several GiB of result artifacts per suite, with the `benchmarks` artifact alone observed at around 8 GiB. Keeping these artifacts for multiple days can quickly exhaust the shared GitHub Actions/Packages storage quota. ## Validation - Ran `git diff --check -- .github/workflows/e2e-run.yml`
Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
The failover detection and recovery path is too slow, caused by three compounding issues: * Polling-only detection with high latency The FailoverControlManager relies a periodic scheduled task running every 1 second to detect fenced brokers and trigger failover. There is no event-driven mechanism — when a broker is fenced by the controller, the failover system doesn't know about it until the next polling cycle. AutoMQ#3406 that it attempts to compensate for detection latency with high-frequency polling, but the polling itself is blocking and consumes the controller's shared event queue — which in turn slows down all operations, including the failover task itself. The correct approach, as this optimization does, is to replace high-frequency polling with event-driven triggering and reduce the periodic scan to a low-frequency safety net (60 seconds). * Sequential I/O during WAL recovery In `S3Storage.recover0()`, `streamManager.getOpeningStreams()` (a remote call) and `deltaWAL.recover()`, `deltaWAL.reset` and `closeStreams` can be executed in parallel instead of sequentially. * Batch delay in FAILOVER WAL write mode During failover recovery, the only write to the WAL is a fake record from `trim()` to persist the trimOffset. In `DefaultWriter`, this record is subject to the normal batching logic (default batchInterval = 250ms), meaning the trim-offset-persisting record sits in the batch for up to 250ms before being uploaded. This delay directly adds to the total failover recovery time.
) - replace per-stream pending append/fetch timestamp deque tracking with broker-level stalled request trackers - add a utils PendingRequestTracker that reports only requests pending beyond the latency threshold - preserve existing pending latency metric names while changing the supplier to broker-level tracking
## Summary - replace the custom `fields/s` TableTopic FPS unit with standard rate unit `1/s` - define the instrument as `kafka_tabletopic_fields`, producing final Prometheus metric `kafka_tabletopic_fields_per_second` - add a regression check for Prometheus metric-name conversion ## Context Runtime 5.5.0 exports `kafka_tabletopic_fps_fields/s` because the previous OTel metric used custom unit `fields/s`. The exporter should not grow one-off mappings for custom business units. Using `name=kafka_tabletopic_fields` with `unit=1/s` follows the existing Prometheus-compatible unit rules and avoids the redundant `fps_fields_per_second` name. Dashboard compatibility is handled in opsbox and automq-labs by matching legacy `kafka_tabletopic_fps_fields/s`, interim `kafka_tabletopic_fps_fields_per_second`, and final `kafka_tabletopic_fields_per_second`. ## Verification - `./gradlew automq-metrics:test --tests com.automq.opentelemetry.exporter.s3.PrometheusUtilsTest` - `./gradlew --build-cache rat checkstyleMain checkstyleTest spotlessJavaCheck`
## Summary - Cherry-pick AutoMQ#3354 to main - Source PR: AutoMQ#3354 - Cherry-picked commit: 89b5f19 ## Test - ./gradlew :connect:runtime:test --tests org.apache.kafka.connect.cli.ConnectStandaloneTest
AutoMQ ListOffset requests can be slow when timestamp lookups need to read sparse data from object storage. A single request may include many partitions, but the existing handling performs the per-partition `fetchOffsetForTimestamp` work sequentially, so the request latency grows with the number of partitions and slow S3-backed lookups can block the request path for a long time. Under overload, the existing path also has no ListOffset-specific concurrency protection, so many expensive partition timestamp lookups can accumulate and amplify latency for later requests. This PR parallelizes ListOffset handling in `KafkaApis` while keeping the change scoped with AutoMQ inject markers: - Runs ListOffset v1+ per-partition `replicaManager.fetchOffsetForTimestamp` calls asynchronously and combines the partition responses with `CompletableFuture`. - Uses dedicated ListOffset executors so expensive timestamp lookups are isolated from the request thread. - Separates fast timestamp requests (`LATEST_TIMESTAMP`, `EARLIEST_TIMESTAMP`, `EARLIEST_LOCAL_TIMESTAMP`) from other timestamp lookups with different executors. - Keeps `MAX_TIMESTAMP` behavior unchanged. - Adds an inflight partition counter and a limit based on the ListOffset thread pool size. When the limit is exceeded, new ListOffset requests fail fast with a retriable overload response (`REQUEST_TIMED_OUT`) instead of queueing unbounded expensive work. - Moves the previous `ElasticKafkaApis` ListOffset delegation back into `KafkaApis`, reducing the AutoMQ-specific override surface. Signed-off-by: Robin Han <hanxvdovehx@gmail.com>
- Move low-level multipart APIs to the `ObjectStorage` interface so wrappers can participate in multipart routing. - Let `MultiPartWriter` fall back to 32 MiB range-read/upload chunks when source and target bucket ids differ. - Return the concrete writer bucket id from `ObjectStorage.write()` and cover cross-bucket copy chunking with a unit test.
- make WriteOptions default bucket id use an explicit UNSET_BUCKET sentinel - update writer tests to set target bucket explicitly when same-bucket copy semantics are expected
sachin428z
requested review from
1sonofqiu,
Gezi-lzq,
superhx and
woshigaopp
as code owners
|
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
feat(metrics): support dynamic metric configuration
fix(metrics): correct dynamic metric validation
docs: improve README for dynamic metrics
refactor(metrics): simplify metric configuration
chore: update metric documentation