This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 5b67e6f120e feat: add configurable metric sampling to the Kafka
emitter (#20559)
5b67e6f120e is described below
commit 5b67e6f120edf77c1b2ec02fd0e383c1f79b5488
Author: Nagarjunanm <[email protected]>
AuthorDate: Wed Oct 21 18:03:27 2026 +0530
feat: add configurable metric sampling to the Kafka emitter (#20559)
* feat: add configurable metric sampling to the Kafka emitter
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
* fix: advance the sampling summary window every interval
The emitted-metric baseline only moved forward in intervals that sampled
something. After a quiet interval, the next summary reported emitted and
total counts covering several intervals instead of the past five minutes.
Move the baseline forward on every status check, and keep logging the
summary only when the interval sampled events. Adds a test for this case.
* perf: decide on metric sampling before serializing the event
Sampled-out metrics were still converted to a map, given extra dimensions,
serialized to JSON and encoded as UTF-8 before being dropped. At low
sampling rates that is wasted CPU and allocation on most metrics, and a
sampled-out event with an unserializable dimension was counted as invalid
instead of sampled.
Check shouldSampleOut() first and return early, so only metrics that will
be queued are serialized. Adds a test for this case.
* refactor: rename sampledNodeTypes to sampledServiceTypes; address review
- Rename druid.emitter.kafka.sampledNodeTypes to sampledServiceTypes, to
match the "service" name used by ServiceMetricEvent.
- Docs: state that samplingRate is an integer from 0 to 100.
- Tests: fix a Javadoc that said rate 10 while the test uses 0, and remove
testShouldSampleOut_noSamplingWhenSampledMetricsEmpty. It used rate 100,
so it could not tell which setting kept the event, and its name described
behaviour the feature doesn't have: an empty sampledMetrics samples every
metric. Both real cases are already covered by
testShouldSampleOut_rate100OverridesAllFilters and
testShouldSampleOut_metricOnlyFilterNonMatchingMetric.
---
.../extensions-contrib/kafka-emitter.md | 27 ++
.../apache/druid/emitter/kafka/KafkaEmitter.java | 102 +++++
.../druid/emitter/kafka/KafkaEmitterConfig.java | 60 ++-
.../emitter/kafka/KafkaEmitterConfigTest.java | 148 ++++++-
.../druid/emitter/kafka/KafkaEmitterTest.java | 474 ++++++++++++++++++++-
5 files changed, 808 insertions(+), 3 deletions(-)
diff --git a/docs/development/extensions-contrib/kafka-emitter.md
b/docs/development/extensions-contrib/kafka-emitter.md
index 772c0ff405c..c08bcecad21 100644
--- a/docs/development/extensions-contrib/kafka-emitter.md
+++ b/docs/development/extensions-contrib/kafka-emitter.md
@@ -49,6 +49,9 @@ All the configuration parameters for the Kafka emitter are
under `druid.emitter.
| `druid.emitter.kafka.extra.dimensions` | Optional JSON configuration to
specify a map of extra string dimensions for the events emitted. These can help
make groups in your monitoring environment. | no | none |
| `druid.emitter.kafka.producer.hiddenProperties` | JSON configuration to
specify sensitive Kafka producer properties such as username and password.
This property accepts a
[DynamicConfigProvider](../../operations/dynamic-config-provider.md)
implementation. | no | none |
| `druid.emitter.kafka.producer.shutdownTimeout` | Duration in milliseconds
the Kafka producer waits for pending requests to finish before shutting down. |
no | Long.MAX_VALUE |
+| `druid.emitter.kafka.samplingRate` | Integer from 0 to 100: the percentage
of matching metric events to keep. `100` keeps every event; `0` drops every
matching event. Decimal values aren't supported. See [Metric
sampling](#metric-sampling). | no | 100 |
+| `druid.emitter.kafka.sampledMetrics` | JSON array of metric names to sample,
such as `["query/time"]`. Names must match exactly. If empty, sampling applies
to all metrics. | no | `[]` |
+| `druid.emitter.kafka.sampledServiceTypes` | JSON array of services to
sample, such as `["druid/broker"]`. If empty, sampling applies to all services.
| no | `[]` |
### Example
@@ -64,3 +67,27 @@
druid.emitter.kafka.extra.dimensions={"region":"us-east-1","environment":"prePro
druid.emitter.kafka.producer.hiddenProperties={"config":{"sasl.jaas.config":
"org.apache.kafka.common.security.plain.PlainLoginModule required
username=\\"KV...NI\\" password=\\"gA3...n6a/\\";"}}
```
+### Metric sampling
+
+High-traffic clusters can emit a large volume of metric events. Metrics such
as `query/time` are emitted for every query. To reduce Kafka produce volume,
network traffic, and storage, the Kafka emitter can drop a percentage of metric
events before they reach the Kafka producer.
+
+Sampling works as follows:
+
+- Only metric events are sampled. Alerts, request logs, and segment metadata
events are always emitted.
+- A metric event is sampled only if its metric name is in `sampledMetrics` and
its service is in `sampledServiceTypes`. An empty list matches everything.
+- Metric events that don't match are always emitted.
+- If a matching event has an `id` dimension, such as the query ID, the
decision to keep or drop it is based on a hash of that ID. Every metric for the
same query is then kept or dropped together. Events without an `id` dimension
are kept at random.
+
+Every five minutes, the emitter logs how many metric events it emitted and how
many it dropped because of sampling.
+
+The following example keeps 10% of query metrics from Brokers:
+
+```
+druid.emitter.kafka.samplingRate=10
+druid.emitter.kafka.sampledMetrics=["query/time", "query/bytes",
"sqlQuery/time"]
+druid.emitter.kafka.sampledServiceTypes=["druid/broker"]
+```
+
+:::info
+Sampling changes absolute counts. When you count sampled events or sum their
values, scale the result by `100 / samplingRate` to estimate the totals.
Distributions such as percentiles don't need scaling.
+:::
diff --git
a/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java
b/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java
index b647e425ebc..0698b94d9c3 100644
---
a/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java
+++
b/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitter.java
@@ -46,6 +46,7 @@ import java.util.Properties;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -66,6 +67,10 @@ public class KafkaEmitter implements Emitter
private final AtomicLong previousRequestLost;
private final AtomicLong previousSegmentMetadataLost;
private final AtomicLong previousInvalidLost;
+ private final AtomicLong metricSampled;
+ private final AtomicLong previousMetricSampled;
+ private final AtomicLong metricEmitted;
+ private final AtomicLong previousMetricEmitted;
private final KafkaEmitterConfig config;
private final Producer<String, String> producer;
@@ -103,6 +108,10 @@ public class KafkaEmitter implements Emitter
this.previousRequestLost = new AtomicLong(0);
this.previousSegmentMetadataLost = new AtomicLong(0);
this.previousInvalidLost = new AtomicLong(0);
+ this.metricSampled = new AtomicLong(0);
+ this.previousMetricSampled = new AtomicLong(0);
+ this.metricEmitted = new AtomicLong(0);
+ this.previousMetricEmitted = new AtomicLong(0);
}
private Callback setProducerCallback(AtomicLong lostCounter)
@@ -175,6 +184,21 @@ public class KafkaEmitter implements Emitter
currentSegmentMetadataLost -
previousSegmentMetadataLost.getAndSet(currentSegmentMetadataLost);
long deltaInvalidLost = currentInvalidLost -
previousInvalidLost.getAndSet(currentInvalidLost);
+ long currentMetricSampled = metricSampled.get();
+ long deltaMetricSampled = currentMetricSampled -
previousMetricSampled.getAndSet(currentMetricSampled);
+ // Advance the emitted baseline every interval, even when nothing was
sampled, so that each summary
+ // only covers the latest interval.
+ long currentMetricEmitted = metricEmitted.get();
+ long deltaMetricEmitted = currentMetricEmitted -
previousMetricEmitted.getAndSet(currentMetricEmitted);
+
+ if (deltaMetricSampled > 0) {
+ long deltaTotal = deltaMetricSampled + deltaMetricEmitted;
+ log.info(
+ "Metrics sampling for past %d minutes: total=[%d], emitted=[%d],
sampledOut=[%d], cumulativeSampledOut=[%d]",
+ DEFAULT_SEND_LOST_INTERVAL_MINUTES,
+ deltaTotal, deltaMetricEmitted, deltaMetricSampled,
currentMetricSampled);
+ }
+
// Refrain from clogging logs when there are no lost messages.
if (deltaMetricLost + deltaAlertLost + deltaRequestLost +
deltaSegmentMetadataLost + deltaInvalidLost > 0) {
log.info(
@@ -229,6 +253,11 @@ public class KafkaEmitter implements Emitter
public void emit(final Event event)
{
if (event != null) {
+ // Decide on sampling first, so sampled-out metrics skip serialization
entirely.
+ if (event instanceof ServiceMetricEvent &&
shouldSampleOut((ServiceMetricEvent) event)) {
+ metricSampled.incrementAndGet();
+ return;
+ }
try {
EventMap map = event.toMap();
map = addExtraDimensionsToEvent(map);
@@ -244,6 +273,8 @@ public class KafkaEmitter implements Emitter
if (event instanceof ServiceMetricEvent) {
if (!eventTypes.contains(EventType.METRICS) ||
!metricQueue.offer(objectContainer)) {
metricLost.incrementAndGet();
+ } else {
+ metricEmitted.incrementAndGet();
}
} else if (event instanceof AlertEvent) {
if (!eventTypes.contains(EventType.ALERTS) ||
!alertQueue.offer(objectContainer)) {
@@ -268,6 +299,55 @@ public class KafkaEmitter implements Emitter
}
}
+ /**
+ * Determines whether a {@link ServiceMetricEvent} should be sampled out
(i.e., dropped).
+ * <p>
+ * Sampling is skipped (event is always emitted) when:
+ * <ul>
+ * <li>{@code samplingRate} is 100 (no sampling)</li>
+ * <li>The metric name is null</li>
+ * <li>The metric name is not in the configured {@code sampledMetrics} set
(exact match only)</li>
+ * <li>The event's service is not in the configured {@code
sampledServiceTypes} set (when configured)</li>
+ * </ul>
+ * When sampling applies, if the event has an {@code id} user dimension,
consistent hash-based
+ * sampling is used so that the same id always gets the same sampling
decision. Otherwise,
+ * random sampling is used.
+ *
+ * @return true if the event should be dropped, false if it should be emitted
+ */
+ @VisibleForTesting
+ boolean shouldSampleOut(ServiceMetricEvent serviceMetricEvent)
+ {
+ int samplingRate = config.getSamplingRate();
+ Set<String> sampledMetrics = config.getSampledMetrics();
+ Set<String> sampledServiceTypes = config.getSampledServiceTypes();
+
+ // No sampling when rate is 100 or metric is null
+ if (samplingRate == 100
+ || serviceMetricEvent.getMetric() == null) {
+ return false;
+ }
+
+ // If sampledMetrics is configured, the metric must be in the set
+ if (!sampledMetrics.isEmpty() &&
!sampledMetrics.contains(serviceMetricEvent.getMetric())) {
+ return false;
+ }
+
+ // If sampledServiceTypes is configured, the service must be in the set
+ if (!sampledServiceTypes.isEmpty()
+ && (serviceMetricEvent.getService() == null
+ ||
!sampledServiceTypes.contains(serviceMetricEvent.getService()))) {
+ return false;
+ }
+
+ // Use consistent hashing on "id" dimension if available, otherwise random
+ Object id = serviceMetricEvent.getUserDims().get("id");
+ if (id == null) {
+ return ThreadLocalRandom.current().nextInt(100) >= samplingRate;
+ }
+ return Math.abs(id.hashCode() % 100) >= samplingRate;
+ }
+
private EventMap addExtraDimensionsToEvent(EventMap map)
{
if (config.getClusterName() != null || config.getExtraDimensions() !=
null) {
@@ -322,6 +402,16 @@ public class KafkaEmitter implements Emitter
return segmentMetadataLost.get();
}
+ public long getMetricSampledCount()
+ {
+ return metricSampled.get();
+ }
+
+ public long getMetricEmittedCount()
+ {
+ return metricEmitted.get();
+ }
+
@VisibleForTesting
protected long getPreviousMetricLostCount()
{
@@ -351,4 +441,16 @@ public class KafkaEmitter implements Emitter
{
return previousInvalidLost.get();
}
+
+ @VisibleForTesting
+ protected long getPreviousMetricSampledCount()
+ {
+ return previousMetricSampled.get();
+ }
+
+ @VisibleForTesting
+ protected long getPreviousMetricEmittedCount()
+ {
+ return previousMetricEmitted.get();
+ }
}
diff --git
a/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitterConfig.java
b/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitterConfig.java
index a690637c27d..f700a58fcd2 100644
---
a/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitterConfig.java
+++
b/extensions-contrib/kafka-emitter/src/main/java/org/apache/druid/emitter/kafka/KafkaEmitterConfig.java
@@ -33,6 +33,7 @@ import org.apache.kafka.clients.producer.ProducerConfig;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
@@ -83,6 +84,14 @@ public class KafkaEmitterConfig
private final DynamicConfigProvider<String> kafkaProducerSecrets;
@JsonProperty("producer.shutdownTimeout")
private final Long shutdownTimeout;
+ @JsonProperty("samplingRate")
+ private final int samplingRate;
+ @JsonProperty("sampledMetrics")
+ @Nonnull
+ private final Set<String> sampledMetrics;
+ @JsonProperty("sampledServiceTypes")
+ @Nonnull
+ private final Set<String> sampledServiceTypes;
@JsonCreator
public KafkaEmitterConfig(
@@ -96,7 +105,10 @@ public class KafkaEmitterConfig
@Nullable @JsonProperty("extra.dimensions") Map<String, String>
extraDimensions,
@JsonProperty("producer.config") @Nullable Map<String, String>
kafkaProducerConfig,
@JsonProperty("producer.hiddenProperties") @Nullable
DynamicConfigProvider<String> kafkaProducerSecrets,
- @JsonProperty("producer.shutdownTimeout") @Nullable Long shutdownTimeout
+ @JsonProperty("producer.shutdownTimeout") @Nullable Long shutdownTimeout,
+ @JsonProperty("samplingRate") @Nullable Integer samplingRate,
+ @JsonProperty("sampledMetrics") @Nullable Set<String> sampledMetrics,
+ @JsonProperty("sampledServiceTypes") @Nullable Set<String>
sampledServiceTypes
)
{
this.eventTypes = maybeUpdateEventTypes(eventTypes, requestTopic);
@@ -143,6 +155,17 @@ public class KafkaEmitterConfig
this.kafkaProducerConfig = kafkaProducerConfig == null ? ImmutableMap.of()
: kafkaProducerConfig;
this.kafkaProducerSecrets = kafkaProducerSecrets == null ? new
MapStringDynamicConfigProvider(ImmutableMap.of()) : kafkaProducerSecrets;
this.shutdownTimeout = Configs.valueOrDefault(shutdownTimeout,
Long.MAX_VALUE);
+ this.samplingRate = Configs.valueOrDefault(samplingRate, 100);
+ if (this.samplingRate < 0 || this.samplingRate > 100) {
+ throw DruidException.forPersona(DruidException.Persona.OPERATOR)
+ .ofCategory(DruidException.Category.INVALID_INPUT)
+ .build(
+ "druid.emitter.kafka.samplingRate must be
between 0 and 100, got [%d].",
+ this.samplingRate
+ );
+ }
+ this.sampledMetrics = sampledMetrics == null ? Collections.emptySet() :
sampledMetrics;
+ this.sampledServiceTypes = sampledServiceTypes == null ?
Collections.emptySet() : sampledServiceTypes;
}
@Nonnull
@@ -228,6 +251,26 @@ public class KafkaEmitterConfig
return shutdownTimeout;
}
+ @JsonProperty
+ public int getSamplingRate()
+ {
+ return samplingRate;
+ }
+
+ @JsonProperty
+ @Nonnull
+ public Set<String> getSampledMetrics()
+ {
+ return sampledMetrics;
+ }
+
+ @JsonProperty
+ @Nonnull
+ public Set<String> getSampledServiceTypes()
+ {
+ return sampledServiceTypes;
+ }
+
@Override
public boolean equals(Object o)
{
@@ -273,6 +316,15 @@ public class KafkaEmitterConfig
if (!getShutdownTimeout().equals(that.getShutdownTimeout())) {
return false;
}
+ if (getSamplingRate() != that.getSamplingRate()) {
+ return false;
+ }
+ if (!getSampledMetrics().equals(that.getSampledMetrics())) {
+ return false;
+ }
+ if (!getSampledServiceTypes().equals(that.getSampledServiceTypes())) {
+ return false;
+ }
return
getKafkaProducerSecrets().getConfig().equals(that.getKafkaProducerSecrets().getConfig());
}
@@ -289,6 +341,9 @@ public class KafkaEmitterConfig
result = 31 * result + (getExtraDimensions() != null ?
getExtraDimensions().hashCode() : 0);
result = 31 * result + getKafkaProducerConfig().hashCode();
result = 31 * result + getShutdownTimeout().hashCode();
+ result = 31 * result + Integer.hashCode(getSamplingRate());
+ result = 31 * result + getSampledMetrics().hashCode();
+ result = 31 * result + getSampledServiceTypes().hashCode();
result = 31 * result + getKafkaProducerSecrets().getConfig().hashCode();
return result;
}
@@ -308,6 +363,9 @@ public class KafkaEmitterConfig
", producer.config=" + kafkaProducerConfig + '\'' +
", producer.hiddenProperties=" + kafkaProducerSecrets +
", producer.shutdownTimeout=" + shutdownTimeout +
+ ", samplingRate=" + samplingRate +
+ ", sampledMetrics=" + sampledMetrics +
+ ", sampledServiceTypes=" + sampledServiceTypes +
'}';
}
}
diff --git
a/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterConfigTest.java
b/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterConfigTest.java
index 6951be915e4..98ffa4f5433 100644
---
a/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterConfigTest.java
+++
b/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterConfigTest.java
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.exc.ValueInstantiationException;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import org.apache.druid.error.DruidException;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.metadata.DynamicConfigProvider;
@@ -57,7 +58,10 @@ public class KafkaEmitterConfigTest
ImmutableMap.<String, String>builder()
.put("testKey", "testValue").build(),
DEFAULT_PRODUCER_SECRETS,
- 50L
+ 50L,
+ null,
+ null,
+ null
);
String kafkaEmitterConfigString =
MAPPER.writeValueAsString(kafkaEmitterConfig);
KafkaEmitterConfig kafkaEmitterConfigExpected =
MAPPER.readerFor(KafkaEmitterConfig.class)
@@ -80,6 +84,9 @@ public class KafkaEmitterConfigTest
ImmutableMap.<String, String>builder()
.put("testKey", "testValue").build(),
DEFAULT_PRODUCER_SECRETS,
+ null,
+ null,
+ null,
null
);
String kafkaEmitterConfigString =
MAPPER.writeValueAsString(kafkaEmitterConfig);
@@ -105,6 +112,9 @@ public class KafkaEmitterConfigTest
ImmutableMap.<String, String>builder()
.put("testKey", "testValue").build(),
DEFAULT_PRODUCER_SECRETS,
+ null,
+ null,
+ null,
null
);
String kafkaEmitterConfigString =
MAPPER.writeValueAsString(kafkaEmitterConfig);
@@ -127,6 +137,9 @@ public class KafkaEmitterConfigTest
ImmutableMap.of("env", "preProd"),
null,
null,
+ null,
+ null,
+ null,
null
);
String kafkaEmitterConfigString =
MAPPER.writeValueAsString(kafkaEmitterConfig);
@@ -182,6 +195,9 @@ public class KafkaEmitterConfigTest
null,
null,
null,
+ null,
+ null,
+ null,
null
)
),
@@ -206,6 +222,9 @@ public class KafkaEmitterConfigTest
null,
null,
null,
+ null,
+ null,
+ null,
null
)
),
@@ -230,6 +249,9 @@ public class KafkaEmitterConfigTest
null,
null,
null,
+ null,
+ null,
+ null,
null
)
),
@@ -254,6 +276,9 @@ public class KafkaEmitterConfigTest
null,
null,
null,
+ null,
+ null,
+ null,
null
)
),
@@ -278,6 +303,9 @@ public class KafkaEmitterConfigTest
null,
null,
null,
+ null,
+ null,
+ null,
null
)
),
@@ -302,6 +330,9 @@ public class KafkaEmitterConfigTest
ImmutableMap.<String, String>builder()
.put("testKey", "testValue").build(),
DEFAULT_PRODUCER_SECRETS,
+ null,
+ null,
+ null,
null
);
String kafkaEmitterConfigString =
MAPPER.writeValueAsString(kafkaEmitterConfig);
@@ -310,6 +341,121 @@ public class KafkaEmitterConfigTest
Assertions.assertEquals(Long.MAX_VALUE, (long)
kafkaEmitterConfigExpected.getShutdownTimeout());
}
+ @Test
+ public void testSamplingDefaults()
+ {
+ KafkaEmitterConfig config = newSamplingConfig(null, null, null);
+ Assertions.assertEquals(100, config.getSamplingRate());
+ Assertions.assertEquals(Collections.emptySet(),
config.getSampledMetrics());
+ Assertions.assertEquals(Collections.emptySet(),
config.getSampledServiceTypes());
+ }
+
+ @Test
+ public void testSerDeserSamplingConfig() throws IOException
+ {
+ KafkaEmitterConfig config = newSamplingConfig(
+ 10,
+ ImmutableSet.of("query/time", "query/bytes"),
+ ImmutableSet.of("druid/broker")
+ );
+ KafkaEmitterConfig roundTripped =
MAPPER.readerFor(KafkaEmitterConfig.class)
+
.readValue(MAPPER.writeValueAsString(config));
+ Assertions.assertEquals(config, roundTripped);
+ Assertions.assertEquals(config.hashCode(), roundTripped.hashCode());
+ Assertions.assertEquals(10, roundTripped.getSamplingRate());
+ Assertions.assertEquals(ImmutableSet.of("query/time", "query/bytes"),
roundTripped.getSampledMetrics());
+ Assertions.assertEquals(ImmutableSet.of("druid/broker"),
roundTripped.getSampledServiceTypes());
+ }
+
+ @Test
+ public void testDeserSamplingConfigFromPropertyNames() throws IOException
+ {
+ KafkaEmitterConfig config = MAPPER.readValue(
+ "{\"bootstrap.servers\": \"localhost:9092\","
+ + " \"metric.topic\": \"metrics\","
+ + " \"alert.topic\": \"alerts\","
+ + " \"samplingRate\": 25,"
+ + " \"sampledMetrics\": [\"query/time\"],"
+ + " \"sampledServiceTypes\": [\"druid/broker\",
\"druid/historical\"]}",
+ KafkaEmitterConfig.class
+ );
+ Assertions.assertEquals(25, config.getSamplingRate());
+ Assertions.assertEquals(ImmutableSet.of("query/time"),
config.getSampledMetrics());
+ Assertions.assertEquals(ImmutableSet.of("druid/broker",
"druid/historical"), config.getSampledServiceTypes());
+ }
+
+ @Test
+ public void testSamplingConfigsAffectEquality()
+ {
+ KafkaEmitterConfig base = newSamplingConfig(10,
ImmutableSet.of("query/time"), ImmutableSet.of("druid/broker"));
+ Assertions.assertEquals(
+ base,
+ newSamplingConfig(10, ImmutableSet.of("query/time"),
ImmutableSet.of("druid/broker"))
+ );
+ Assertions.assertNotEquals(
+ base,
+ newSamplingConfig(20, ImmutableSet.of("query/time"),
ImmutableSet.of("druid/broker"))
+ );
+ Assertions.assertNotEquals(
+ base,
+ newSamplingConfig(10, ImmutableSet.of("query/bytes"),
ImmutableSet.of("druid/broker"))
+ );
+ Assertions.assertNotEquals(
+ base,
+ newSamplingConfig(10, ImmutableSet.of("query/time"),
ImmutableSet.of("druid/historical"))
+ );
+ }
+
+ @Test
+ public void testSamplingRateBoundariesAccepted()
+ {
+ Assertions.assertEquals(0, newSamplingConfig(0, null,
null).getSamplingRate());
+ Assertions.assertEquals(100, newSamplingConfig(100, null,
null).getSamplingRate());
+ }
+
+ @Test
+ public void testSamplingRateOutOfRangeRejected()
+ {
+ for (int invalidRate : new int[]{-1, 101}) {
+ DruidException exception = Assertions.assertThrows(
+ DruidException.class,
+ () -> newSamplingConfig(invalidRate, null, null)
+ );
+ Assertions.assertAll(
+ () -> Assertions.assertEquals(DruidException.Persona.OPERATOR,
exception.getTargetPersona()),
+ () -> Assertions.assertEquals(DruidException.Category.INVALID_INPUT,
exception.getCategory()),
+ () -> Assertions.assertEquals(
+ "druid.emitter.kafka.samplingRate must be between 0 and 100, got
[" + invalidRate + "].",
+ exception.getMessage()
+ )
+ );
+ }
+ }
+
+ private static KafkaEmitterConfig newSamplingConfig(
+ Integer samplingRate,
+ Set<String> sampledMetrics,
+ Set<String> sampledServiceTypes
+ )
+ {
+ return new KafkaEmitterConfig(
+ "localhost:9092",
+ null,
+ "metrics",
+ "alerts",
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ samplingRate,
+ sampledMetrics,
+ sampledServiceTypes
+ );
+ }
+
private void assertOperatorException(final DruidException exception, final
String expectedMessage)
{
Assertions.assertAll(
diff --git
a/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterTest.java
b/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterTest.java
index 0f6f91df1c6..34c3bbd0a10 100644
---
a/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterTest.java
+++
b/extensions-contrib/kafka-emitter/src/test/java/org/apache/druid/emitter/kafka/KafkaEmitterTest.java
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.StringUtils;
@@ -52,6 +53,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -151,6 +153,9 @@ public class KafkaEmitterTest
ImmutableMap.of("clusterId", "cluster-101"),
null,
null,
+ null,
+ null,
+ null,
null
);
@@ -206,6 +211,9 @@ public class KafkaEmitterTest
ImmutableMap.of("clusterId", "cluster-101", "env", "staging"),
null,
null,
+ null,
+ null,
+ null,
null
);
@@ -257,6 +265,9 @@ public class KafkaEmitterTest
null,
null,
null,
+ null,
+ null,
+ null,
null
);
@@ -307,6 +318,9 @@ public class KafkaEmitterTest
null,
null,
null,
+ null,
+ null,
+ null,
null
);
@@ -383,6 +397,9 @@ public class KafkaEmitterTest
null,
null,
null,
+ null,
+ null,
+ null,
null
);
@@ -432,7 +449,10 @@ public class KafkaEmitterTest
null,
null,
"cluster-102",
- ImmutableMap.of("clusterName", "cluster-101", "env", "staging"), //
clusterName again, extraDimensions should take precedence
+ ImmutableMap.of("clusterName", "cluster-101", "env", "staging"),
+ null,
+ null,
+ null,
null,
null,
null
@@ -530,6 +550,9 @@ public class KafkaEmitterTest
extraDimensions,
ImmutableMap.of(ProducerConfig.BUFFER_MEMORY_CONFIG,
String.valueOf(totalBufferSize)),
null,
+ null,
+ null,
+ null,
null
);
@@ -672,4 +695,453 @@ public class KafkaEmitterTest
return "testFeed";
}
}
+
+ // ==================== Sampling Tests ====================
+
+ @Test
+ public void testShouldSampleOut_noSamplingWhenRateIs100()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(100,
ImmutableSet.of("query/time"), null));
+ ServiceMetricEvent event =
ServiceMetricEvent.builder().setMetric("query/time", 100).build("svc", "host");
+ Assertions.assertFalse(emitter.shouldSampleOut(event));
+ }
+
+ /**
+ * When both sampledMetrics and sampledServiceTypes are empty, every metric
is eligible for sampling.
+ * samplingRate=0 makes the result deterministic: every metric is dropped.
+ */
+ @Test
+ public void testShouldSampleOut_samplesAllWhenRateSetButFiltersEmpty()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(0, null,
null));
+ ServiceMetricEvent event =
ServiceMetricEvent.builder().setMetric("query/time", 100).build("svc", "host");
+ Assertions.assertTrue(emitter.shouldSampleOut(event));
+ }
+
+ @Test
+ public void testShouldSampleOut_noSamplingForNonMatchingMetric()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(
+ 10,
+ ImmutableSet.of("query/time", "query/bytes"),
+ null
+ ));
+ // "jvm/mem/used" is not in sampledMetrics, so should never be sampled out
+ ServiceMetricEvent event =
ServiceMetricEvent.builder().setMetric("jvm/mem/used", 100).build("svc",
"host");
+ Assertions.assertFalse(emitter.shouldSampleOut(event));
+ }
+
+ @Test
+ public void testShouldSampleOut_samplesMatchingMetricWithZeroRate()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(0,
ImmutableSet.of("query/time"), null));
+ ServiceMetricEvent event =
ServiceMetricEvent.builder().setMetric("query/time", 100).build("svc", "host");
+ // With rate=0, all matching metrics should be sampled out
+ Assertions.assertTrue(emitter.shouldSampleOut(event));
+ }
+
+ @Test
+ public void testShouldSampleOut_consistentHashOnId()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(50,
ImmutableSet.of("query/time"), null));
+
+ // Build event with an "id" dimension
+ ServiceMetricEvent event = ServiceMetricEvent.builder()
+ .setDimension("id", "query-abc-123")
+ .setMetric("query/time", 100)
+ .build("svc", "host");
+
+ // Consistent hashing: same id should always produce the same result
+ boolean firstResult = emitter.shouldSampleOut(event);
+ for (int i = 0; i < 100; i++) {
+ Assertions.assertEquals(
+ firstResult,
+ emitter.shouldSampleOut(event),
+ "Consistent hashing should give same result for same id"
+ );
+ }
+ }
+
+ // ==================== Service Type Sampling Tests ====================
+
+ /**
+ * samplingRate=0, sampledMetrics=["query/time"],
sampledServiceTypes=["druid/broker"]
+ * Event from druid/broker with query/time → should be sampled out
+ */
+ @Test
+ public void testShouldSampleOut_matchingMetricAndServiceType()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(
+ 0,
+ ImmutableSet.of("query/time"),
+ ImmutableSet.of("druid/broker")
+ ));
+ ServiceMetricEvent event = ServiceMetricEvent.builder()
+ .setMetric("query/time", 100)
+ .build("druid/broker", "host");
+ Assertions.assertTrue(emitter.shouldSampleOut(event));
+ }
+
+ /**
+ * samplingRate=0, sampledMetrics=["query/time"],
sampledServiceTypes=["druid/broker"]
+ * Event from druid/historical with query/time → should NOT be sampled
(wrong service type)
+ */
+ @Test
+ public void testShouldSampleOut_matchingMetricButWrongServiceType()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(
+ 0,
+ ImmutableSet.of("query/time"),
+ ImmutableSet.of("druid/broker")
+ ));
+ ServiceMetricEvent event = ServiceMetricEvent.builder()
+ .setMetric("query/time", 100)
+ .build("druid/historical", "host");
+ Assertions.assertFalse(emitter.shouldSampleOut(event));
+ }
+
+ /**
+ * samplingRate=0, sampledMetrics=empty, sampledServiceTypes=["druid/broker"]
+ * Event from druid/broker with any metric → should be sampled out (service
type filter only)
+ */
+ @Test
+ public void testShouldSampleOut_serviceTypeOnlyFilter()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(0, null,
ImmutableSet.of("druid/broker")));
+ ServiceMetricEvent event = ServiceMetricEvent.builder()
+ .setMetric("jvm/mem/used", 100)
+ .build("druid/broker", "host");
+ Assertions.assertTrue(emitter.shouldSampleOut(event));
+ }
+
+ /**
+ * samplingRate=0, sampledMetrics=empty, sampledServiceTypes=["druid/broker"]
+ * Event from druid/historical → should NOT be sampled (wrong service type)
+ */
+ @Test
+ public void testShouldSampleOut_serviceTypeOnlyFilterWrongServiceType()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(0, null,
ImmutableSet.of("druid/broker")));
+ ServiceMetricEvent event = ServiceMetricEvent.builder()
+ .setMetric("jvm/mem/used", 100)
+ .build("druid/historical", "host");
+ Assertions.assertFalse(emitter.shouldSampleOut(event));
+ }
+
+ /**
+ * samplingRate=0, sampledMetrics=["query/time"], sampledServiceTypes=empty
+ * Event with query/time from any node → should be sampled out (metric
filter only)
+ */
+ @Test
+ public void testShouldSampleOut_metricOnlyFilter()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(0,
ImmutableSet.of("query/time"), null));
+ ServiceMetricEvent event = ServiceMetricEvent.builder()
+ .setMetric("query/time", 100)
+ .build("druid/historical", "host");
+ Assertions.assertTrue(emitter.shouldSampleOut(event));
+ }
+
+ /**
+ * samplingRate=0, sampledMetrics=["query/time"], sampledServiceTypes=empty
+ * Event with jvm/mem/used → should NOT be sampled (metric doesn't match)
+ */
+ @Test
+ public void testShouldSampleOut_metricOnlyFilterNonMatchingMetric()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(0,
ImmutableSet.of("query/time"), null));
+ ServiceMetricEvent event = ServiceMetricEvent.builder()
+ .setMetric("jvm/mem/used", 100)
+ .build("druid/historical", "host");
+ Assertions.assertFalse(emitter.shouldSampleOut(event));
+ }
+
+ /**
+ * samplingRate=100 always means no sampling, even with filters configured
+ */
+ @Test
+ public void testShouldSampleOut_rate100OverridesAllFilters()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(
+ 100,
+ ImmutableSet.of("query/time"),
+ ImmutableSet.of("druid/broker")
+ ));
+ ServiceMetricEvent event = ServiceMetricEvent.builder()
+ .setMetric("query/time", 100)
+ .build("druid/broker", "host");
+ Assertions.assertFalse(emitter.shouldSampleOut(event));
+ }
+
+ /**
+ * samplingRate=0, both filters empty → sample ALL metrics from all nodes
(drop everything)
+ */
+ @Test
+ public void testShouldSampleOut_rate0WithNoFiltersDropsEverything()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(0, null,
null));
+ ServiceMetricEvent event = ServiceMetricEvent.builder()
+ .setMetric("anything/at/all", 100)
+ .build("druid/coordinator", "host");
+ Assertions.assertTrue(emitter.shouldSampleOut(event));
+ }
+
+ /**
+ * samplingRate=0, multiple service types configured
+ */
+ @Test
+ public void testShouldSampleOut_multipleServiceTypes()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(
+ 0,
+ null,
+ ImmutableSet.of("druid/broker", "druid/historical")
+ ));
+
+ // broker → sampled
+ ServiceMetricEvent brokerEvent = ServiceMetricEvent.builder()
+ .setMetric("query/time", 100)
+ .build("druid/broker", "host");
+ Assertions.assertTrue(emitter.shouldSampleOut(brokerEvent));
+
+ // historical → sampled
+ ServiceMetricEvent historicalEvent = ServiceMetricEvent.builder()
+ .setMetric("query/time", 100)
+ .build("druid/historical", "host");
+ Assertions.assertTrue(emitter.shouldSampleOut(historicalEvent));
+
+ // coordinator → NOT sampled (not in set)
+ ServiceMetricEvent coordinatorEvent = ServiceMetricEvent.builder()
+ .setMetric("query/time", 100)
+ .build("druid/coordinator", "host");
+ Assertions.assertFalse(emitter.shouldSampleOut(coordinatorEvent));
+ }
+
+ /**
+ * With 10% sampling and no "id" dimension, roughly 10% of matching events
should be kept.
+ * The bounds are wide enough (about 13 standard deviations) for this check
never to flake.
+ */
+ @Test
+ public void testShouldSampleOut_approximateRetentionRate()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(10,
ImmutableSet.of("query/time"), null));
+ final ServiceMetricEvent event = ServiceMetricEvent.builder()
+
.setMetric("query/time", 100)
+ .build("druid/broker",
"host");
+ final int total = 100_000;
+ int kept = 0;
+ for (int i = 0; i < total; i++) {
+ if (!emitter.shouldSampleOut(event)) {
+ kept++;
+ }
+ }
+ Assertions.assertTrue(kept > 8_800 && kept < 11_200, "Expected about 10%
kept, but kept " + kept + " of " + total);
+ }
+
+ /**
+ * With 50% sampling, events with different "id" values should be split
between kept and dropped,
+ * while each individual id keeps getting the same decision.
+ */
+ @Test
+ public void testShouldSampleOut_hashOnIdSplitsAcrossIds()
+ {
+ final KafkaEmitter emitter = initKafkaEmitter(newSamplingConfig(50,
ImmutableSet.of("query/time"), null));
+ final int total = 10_000;
+ int kept = 0;
+ for (int i = 0; i < total; i++) {
+ final ServiceMetricEvent event = ServiceMetricEvent.builder()
+ .setDimension("id",
"query-" + i)
+
.setMetric("query/time", 100)
+
.build("druid/broker", "host");
+ final boolean dropped = emitter.shouldSampleOut(event);
+ Assertions.assertEquals(dropped, emitter.shouldSampleOut(event));
+ if (!dropped) {
+ kept++;
+ }
+ }
+ Assertions.assertTrue(kept > 4_000 && kept < 6_000, "Expected about 50%
kept, but kept " + kept + " of " + total);
+ }
+
+ /**
+ * End-to-end: with samplingRate=0, metric events never reach the Kafka
producer, while alerts, requests and
+ * segment metadata are still emitted. Sampled-out events are counted as
sampled, not as lost.
+ */
+ @Test
+ @Timeout(value = 10_000, unit = TimeUnit.MILLISECONDS)
+ public void testSampledMetricsAreDroppedBeforeProducerAndOtherEventsAreKept()
+ throws InterruptedException, JsonProcessingException
+ {
+ final KafkaEmitterConfig kafkaEmitterConfig = new KafkaEmitterConfig(
+ "",
+ new HashSet<>(Arrays.asList(KafkaEmitterConfig.EventType.values())),
+ "metrics",
+ "alerts",
+ "requests",
+ "segments",
+ "clusterName",
+ null,
+ null,
+ null,
+ null,
+ 0,
+ null,
+ null
+ );
+ final KafkaEmitter kafkaEmitter = initKafkaEmitter(kafkaEmitterConfig);
+
+ final List<Event> metricEvents = flattenEvents(SERVICE_METRIC_EVENTS);
+ final List<Event> keptEvents = flattenEvents(ALERT_EVENTS,
REQUEST_LOG_EVENTS, SEGMENT_METADATA_EVENTS);
+ final CountDownLatch eventLatch = new CountDownLatch(keptEvents.size());
+
+ final Map<String, List<EventMap>> feedToExpectedEvents =
trackExpectedEventsPerFeed(
+ keptEvents,
+ kafkaEmitterConfig.getClusterName(),
+ kafkaEmitterConfig.getExtraDimensions()
+ );
+ final Map<String, List<EventMap>> feedToActualEvents =
trackActualEventsPerFeed(eventLatch);
+
+ emitEvents(kafkaEmitter, flattenEvents(metricEvents, keptEvents),
eventLatch);
+
+ validateEvents(feedToExpectedEvents, feedToActualEvents);
+ Assertions.assertEquals(metricEvents.size(),
kafkaEmitter.getMetricSampledCount());
+ Assertions.assertEquals(0, kafkaEmitter.getMetricEmittedCount());
+ Assertions.assertEquals(0, kafkaEmitter.getMetricLostCount());
+ Assertions.assertEquals(0, kafkaEmitter.getAlertLostCount());
+ Assertions.assertEquals(0, kafkaEmitter.getRequestLostCount());
+ Assertions.assertEquals(0, kafkaEmitter.getSegmentMetadataLostCount());
+ }
+
+ /**
+ * End-to-end: only the configured metric is sampled; other metrics are
still emitted.
+ */
+ @Test
+ @Timeout(value = 10_000, unit = TimeUnit.MILLISECONDS)
+ public void testOnlyConfiguredMetricsAreSampledEndToEnd() throws
InterruptedException, JsonProcessingException
+ {
+ final KafkaEmitterConfig kafkaEmitterConfig = newSamplingConfig(0,
ImmutableSet.of("query/time"), null);
+ final KafkaEmitter kafkaEmitter = initKafkaEmitter(kafkaEmitterConfig);
+
+ final List<Event> sampledEvents = ImmutableList.of(
+ ServiceMetricEvent.builder().setMetric("query/time",
10).build("druid/broker", "host"),
+ ServiceMetricEvent.builder().setMetric("query/time",
20).build("druid/broker", "host")
+ );
+ final List<Event> keptEvents = ImmutableList.of(
+ ServiceMetricEvent.builder().setMetric("jvm/mem/used",
30).build("druid/broker", "host"),
+ ServiceMetricEvent.builder().setMetric("query/bytes",
40).build("druid/broker", "host")
+ );
+ final CountDownLatch eventLatch = new CountDownLatch(keptEvents.size());
+
+ final Map<String, List<EventMap>> feedToExpectedEvents =
trackExpectedEventsPerFeed(
+ keptEvents,
+ kafkaEmitterConfig.getClusterName(),
+ kafkaEmitterConfig.getExtraDimensions()
+ );
+ final Map<String, List<EventMap>> feedToActualEvents =
trackActualEventsPerFeed(eventLatch);
+
+ emitEvents(kafkaEmitter, flattenEvents(sampledEvents, keptEvents),
eventLatch);
+
+ validateEvents(feedToExpectedEvents, feedToActualEvents);
+ Assertions.assertEquals(sampledEvents.size(),
kafkaEmitter.getMetricSampledCount());
+ Assertions.assertEquals(keptEvents.size(),
kafkaEmitter.getMetricEmittedCount());
+ Assertions.assertEquals(0, kafkaEmitter.getMetricLostCount());
+
+ // Exercises the periodic sampling summary log path.
+ kafkaEmitter.logLostMessagesStatus();
+ }
+
+ /**
+ * The sampling summary must only cover the latest interval. An interval
that emits metrics but samples none
+ * skips the log line, but must still move the emitted-count baseline
forward, so its emissions are not
+ * reported again in a later interval's summary.
+ */
+ @Test
+ public void testSamplingSummaryWindowAdvancesWhenNothingSampled()
+ {
+ final KafkaEmitter kafkaEmitter = initKafkaEmitter(newSamplingConfig(0,
ImmutableSet.of("query/time"), null));
+ final ServiceMetricEvent keptEvent = ServiceMetricEvent.builder()
+
.setMetric("jvm/mem/used", 1)
+
.build("druid/broker", "host");
+ final ServiceMetricEvent sampledEvent = ServiceMetricEvent.builder()
+
.setMetric("query/time", 1)
+
.build("druid/broker", "host");
+
+ // Interval 1: two metrics emitted, none sampled out.
+ kafkaEmitter.emit(keptEvent);
+ kafkaEmitter.emit(keptEvent);
+ kafkaEmitter.logLostMessagesStatus();
+ Assertions.assertEquals(0, kafkaEmitter.getMetricSampledCount());
+ Assertions.assertEquals(2, kafkaEmitter.getPreviousMetricEmittedCount());
+
+ // Interval 2: one emitted and one sampled out. The window must start from
interval 1's end.
+ kafkaEmitter.emit(keptEvent);
+ kafkaEmitter.emit(sampledEvent);
+ kafkaEmitter.logLostMessagesStatus();
+ Assertions.assertEquals(3, kafkaEmitter.getPreviousMetricEmittedCount());
+ Assertions.assertEquals(1, kafkaEmitter.getPreviousMetricSampledCount());
+ }
+
+ /**
+ * Sampling is decided before the event is serialized, so a sampled-out
metric never pays for serialization.
+ * A dimension that fails to serialize proves it: the sampled-out event is
counted as sampled, not invalid,
+ * while an event that is kept still goes through serialization and is
counted as invalid as before.
+ */
+ @Test
+ public void testSamplingHappensBeforeSerialization()
+ {
+ final KafkaEmitter kafkaEmitter = initKafkaEmitter(newSamplingConfig(0,
ImmutableSet.of("query/time"), null));
+
+ kafkaEmitter.emit(
+ ServiceMetricEvent.builder()
+ .setDimension("bad", new UnserializableValue())
+ .setMetric("query/time", 1)
+ .build("druid/broker", "host")
+ );
+ Assertions.assertEquals(1, kafkaEmitter.getMetricSampledCount());
+ Assertions.assertEquals(0, kafkaEmitter.getInvalidLostCount());
+
+ kafkaEmitter.emit(
+ ServiceMetricEvent.builder()
+ .setDimension("bad", new UnserializableValue())
+ .setMetric("jvm/mem/used", 1)
+ .build("druid/broker", "host")
+ );
+ Assertions.assertEquals(1, kafkaEmitter.getMetricSampledCount());
+ Assertions.assertEquals(1, kafkaEmitter.getInvalidLostCount());
+ }
+
+ /**
+ * A dimension value that Jackson cannot serialize, because its only
property getter throws.
+ */
+ private static class UnserializableValue
+ {
+ @SuppressWarnings("unused")
+ public String getValue()
+ {
+ throw new IllegalStateException("not serializable");
+ }
+ }
+
+ private static KafkaEmitterConfig newSamplingConfig(
+ final Integer samplingRate,
+ final Set<String> sampledMetrics,
+ final Set<String> sampledServiceTypes
+ )
+ {
+ return new KafkaEmitterConfig(
+ "localhost:9092",
+ null,
+ "metrics",
+ "alerts",
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ samplingRate,
+ sampledMetrics,
+ sampledServiceTypes
+ );
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
