Package ai.signalroom.kafka.isotope

Class IsotopeContext

java.lang.Object

ai.signalroom.kafka.isotope.IsotopeContext

public final class IsotopeContext
extends Object

Thread-local handoff for the in-flight isotope between a consumer and a downstream producer in the same processing thread. In a typical consume-then-produce service, the application calls adoptFromRecord(ConsumerRecord) before processing each record so that IsotopeProducerInterceptor can find the inbound trace context and append the next hop instead of starting a new trace. The (record, service) overload additionally emits the stateless isotope.consume.age timer when the Micrometer exporter is enabled.

For bipartite topology visibility, consumers also call recordConsume(ConsumerRecord, String, Producer) to emit a best-effort consume-edge marker to "isotope_consume_edge_markers".

Field Summary

Fields

Modifier and Type	Field	Description
static final String	CONSUME_EVENTS_TOPIC	Default topic that recordConsume(org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>, java.lang.String, org.apache.kafka.clients.producer.Producer<byte[], byte[]>) writes markers to.

Method Summary

Modifier and Type	Method	Description
static Isotope	adoptFromRecord(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> consumerRecord)	Extracts the isotope from the given record's headers (if any) and installs it as the current thread-local context.
static Isotope	adoptFromRecord(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> consumerRecord, String consumerService)	Like adoptFromRecord(ConsumerRecord), but also emits the stateless consume-side isotope.consume.age timer attributing the adoption to consumerService.
static void	clear()
static Isotope	current()
static void	recordConsume(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> consumerRecord, String consumerService, org.apache.kafka.clients.producer.Producer<byte[],byte[]> emitter)	Emits a best-effort consume-edge marker to "isotope_consume_edge_markers" using the default topic name.
static void	recordConsume(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> consumerRecord, String consumerService, org.apache.kafka.clients.producer.Producer<byte[],byte[]> emitter, String topic)	Emits a best-effort consume-edge marker describing the consumption of record by consumerService.
static void	set(Isotope iso)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

CONSUME_EVENTS_TOPIC

public static final String CONSUME_EVENTS_TOPIC

Default topic that recordConsume(org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>, java.lang.String, org.apache.kafka.clients.producer.Producer<byte[], byte[]>) writes markers to.

See Also:

• Constant Field Values

Method Details

current

public static Isotope current()

set

public static void set(Isotope iso)

clear

public static void clear()

adoptFromRecord

public static Isotope adoptFromRecord(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> consumerRecord)

Extracts the isotope from the given record's headers (if any) and installs it as the current thread-local context. Returns the isotope adopted, or null if the record carried no isotope.

Emits no metric — use adoptFromRecord(ConsumerRecord, String) to also record the stateless isotope.consume.age timer.

adoptFromRecord

public static Isotope adoptFromRecord(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> consumerRecord,
 String consumerService)

Like adoptFromRecord(ConsumerRecord), but also emits the stateless consume-side isotope.consume.age timer attributing the adoption to consumerService. The age is now - originTs — how stale the record was when this service adopted it to continue the trace — the adoption-path complement of recordConsume(org.apache.kafka.clients.consumer.ConsumerRecord<?, ?>, java.lang.String, org.apache.kafka.clients.producer.Producer<byte[], byte[]>)'s marker-path isotope.consume.latency.

The thread-local adoption itself is identical to the single-arg overload. The metric is a no-op when the record carries no isotope, consumerService is null, or the exporter is off (IsotopeMetrics.isEnabled()).

recordConsume

public static void recordConsume(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> consumerRecord,
 String consumerService,
 org.apache.kafka.clients.producer.Producer<byte[],byte[]> emitter)

Emits a best-effort consume-edge marker to "isotope_consume_edge_markers" using the default topic name. Equivalent to recordConsume(ConsumerRecord, String, Producer, String) with topic = CONSUME_EVENTS_TOPIC.

recordConsume

public static void recordConsume(org.apache.kafka.clients.consumer.ConsumerRecord<?,?> consumerRecord,
 String consumerService,
 org.apache.kafka.clients.producer.Producer<byte[],byte[]> emitter,
 String topic)

Emits a best-effort consume-edge marker describing the consumption of record by consumerService. The marker is an empty-value record on topic whose headers forward every scalar x-isotope-* header from the inbound record plus a new Isotope.HEADER_CONSUMER_SERVICE naming the consumer.

No-op when the inbound record carries no Isotope.HEADER_TRACE_ID - untagged records don't belong in the bipartite graph.

Fire-and-forget: the send is async with no callback. The Producer lifecycle is the caller's responsibility.

When the Micrometer exporter is enabled (

invalid reference

IsotopeMetrics#start

), this also emits the stateless consume-edge metrics for the marker via IsotopeMetrics.recordConsume(java.lang.String, java.lang.String, java.lang.String, java.lang.String, long) — the consume-side analogue of the produce-side IsotopeProducerInterceptor emission. For terminal consumers (those that did not adopt the trace on this thread, so current() is null) it additionally emits the isotope.consume.age timer that adoption would otherwise miss; when the record was adopted, age was already emitted on the adoption path and is skipped here to avoid a double sample.