Package ai.signalroom.kafka.isotope

Interface IsotopeMetricsSink

public interface IsotopeMetricsSink

The metrics seam between kafka-isotope-core (trace propagation) and any metrics backend. Core emits through this interface via the IsotopeMetrics facade, so adopters who only want propagation never pull a metrics dependency.

The default sink is NoOpMetricsSink (all methods inert, isEnabled() false). The optional kafka-isotope-metrics module provides a Micrometer/Prometheus implementation (PrometheusIsotopeMetrics) and registers it via IsotopeMetrics.register(IsotopeMetricsSink) when its exporter starts.

Method contracts mirror the original meters — see IsotopeMetrics. Implementations are called from Kafka send/consume threads and must be thread-safe.

Method Summary

Modifier and Type	Method	Description
boolean	isEnabled()	True once this sink is actively recording.
void	recordConsume(String pipeline, String originService, String consumerService, String thisTopic, long latencyMs)	Records the consume-edge count and origin→consume latency for one consume marker.
void	recordConsumeAge(String pipeline, String originService, String consumerService, String thisTopic, long ageMs)	Records the consume-side "age" timer (now - originTs) for one consumed record — emitted once on every consuming stage.
void	recordHop(String pipeline, String originService, String thisService, String thisTopic, long latencyMs, int hopCount)	Records the stateless-aggregation metrics for one produced hop: origin→hop latency plus the hop-distribution counter.

Method Details

isEnabled

boolean isEnabled()

True once this sink is actively recording. Core guards emission on this so that, with the no-op sink (or before the exporter starts), the hot path skips building tag values.

recordHop

void recordHop(String pipeline,
 String originService,
 String thisService,
 String thisTopic,
 long latencyMs,
 int hopCount)

Records the stateless-aggregation metrics for one produced hop: origin→hop latency plus the hop-distribution counter.

Parameters:

pipeline - the trace's pipeline (origin-set, forwarded)

originService - the service that originated the trace

thisService - the service producing this hop

thisTopic - the topic this hop is produced to

latencyMs - thisHopTs - originTs; clamp at 0 for skew

hopCount - number of hops accumulated including this one

recordConsume

void recordConsume(String pipeline,
 String originService,
 String consumerService,
 String thisTopic,
 long latencyMs)

Records the consume-edge count and origin→consume latency for one consume marker.

Parameters:

latencyMs - consumeTs - originTs, or < 0 when the record carried no origin timestamp (edge still counted, latency timer skipped)

recordConsumeAge

void recordConsumeAge(String pipeline,
 String originService,
 String consumerService,
 String thisTopic,
 long ageMs)

Records the consume-side "age" timer (now - originTs) for one consumed record — emitted once on every consuming stage.

Parameters:

ageMs - now - originTs; clamp at 0 for skew