feat: add configurable metric sampling to the Kafka emitter

[Nagarjunanm]

Adds optional sampling of ServiceMetricEvents to the kafka-emitter extension,
so operators can drop a percentage of high-volume metrics before they reach the
Kafka producer.

New properties under druid.emitter.kafka:

• samplingRate (0-100, default 100 = no sampling); values outside 0-100 are rejected
• sampledMetrics: exact metric names eligible for sampling (empty = all)
• sampledNodeTypes: services eligible for sampling (empty = all)

Only metric events are sampled; alerts, request logs and segment metadata are
always emitted. When an event has an "id" dimension, the keep/drop decision is
based on a hash of that id, so all metrics for one query are kept or dropped
together; otherwise events are kept at random. The emitter counts sampled and
emitted metric events and logs a summary every five minutes.

Fixes #20246

---

PR title

feat: add configurable metric sampling to the Kafka emitter

PR description

Fixes #20246.

Description

High-traffic clusters emit a very large number of metric events, especially per-query metrics such as
query/time. When those metrics are used for trends and percentiles, keeping all of them costs Kafka
throughput, network and storage without adding much value. Filtering on the consumer side doesn't help,
because the events have already been serialized, produced and stored by then.

This PR adds opt-in sampling to the Kafka emitter. Events are dropped in KafkaEmitter.emit(), before
they're queued for the producer.

New configuration

Property	Default	Description
druid.emitter.kafka.samplingRate	100	Percentage (0-100) of matching metric events to keep. Values outside 0-100 fail at startup with an INVALID_INPUT error.
druid.emitter.kafka.sampledMetrics	[]	Metric names to sample (exact match). Empty means all metrics.
druid.emitter.kafka.sampledNodeTypes	[]	Services to sample, such as druid/broker. Empty means all services.

The default samplingRate of 100 keeps today's behavior.

Behavior

• Only ServiceMetricEvents are sampled. Alerts, request logs and segment metadata are always emitted.
• An event is sampled only if it matches both sampledMetrics and sampledNodeTypes. An empty list matches everything.
• If a matching event has an id dimension (for example the query ID), the decision is based on a hash of
  that ID, so every metric for the same query is kept or dropped together. This keeps per-query metrics
  consistent with each other. Events without an id are kept at random using ThreadLocalRandom.
• Sampled-out events are counted separately from lost events. The emitter logs a sampling summary
  (total, emitted, sampled out) every five minutes, only when something was sampled.

Example

druid.emitter.kafka.samplingRate=10
druid.emitter.kafka.sampledMetrics=["query/time", "query/bytes", "sqlQuery/time"]
druid.emitter.kafka.sampledNodeTypes=["druid/broker"]

Release note

The Kafka emitter can now sample metric events with the new druid.emitter.kafka.samplingRate,
druid.emitter.kafka.sampledMetrics and druid.emitter.kafka.sampledNodeTypes properties. This reduces
Kafka produce volume for high-traffic metrics such as query/time. Sampling is off by default.

---

Key changed/added classes in this PR

• KafkaEmitterConfig
• KafkaEmitter
• KafkaEmitterConfigTest
• KafkaEmitterTest
• docs/development/extensions-contrib/kafka-emitter.md

---

This PR has:

• been self-reviewed.
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.

[FrankChen021]

🟡 Changes recommended

Fix the sampling-summary counter window before merging. When an interval has emitted metrics but no sampled-out metrics, the next summary reports emitted and total values accumulated across multiple five-minute intervals instead of the advertised past-five-minute window.

Reviewed 5 of 5 changed files; findings in 1 file.

Validation: git diff --check 87525a9c6d50474efc94cbc01706fa28c99b1dc1..8f58b5448ce55f86f82fcb4d2c81bd5603c347aa passed. No builds or tests were run.

Severity	Findings
P0	0
P1	0
P2	1
P3	0
Total	1

---

This is an automated review by Codex GPT-5.6-Luna(max)

After addressing the findings or replying to the comments, you can request another review from me to trigger a new automated review.

[FrankChen021]

[P2] Sampling summary retains a stale emitted counter

Finding: If a five-minute interval has emitted metrics but deltaMetricSampled is zero, this branch is skipped and previousMetricEmitted is not advanced. When a later interval samples anything, deltaMetricEmitted includes emissions from all skipped intervals, so the logged total and emitted values no longer describe the past five-minute window and can mislead operators about sampling volume.

Suggestion: Advance previousMetricEmitted on every status check, while keeping the sampling-summary log conditional on whether the current interval sampled any events.