Package ai.signalroom.kafka.isotope

Class Isotope

java.lang.Object

ai.signalroom.kafka.isotope.Isotope

public final class Isotope
extends Object

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	Isotope.Hop

Field Summary

Fields

Modifier and Type	Field	Description
static final String	HEADER_CONSUMER_SERVICE	Consume-side marker header.
static final String	HEADER_HOP_COUNT
static final String	HEADER_KEY
static final String	HEADER_ORIGIN_SERVICE
static final String	HEADER_ORIGIN_TS
static final String	HEADER_PIPELINE
static final String	HEADER_THIS_SERVICE
static final String	HEADER_THIS_TOPIC
static final String	HEADER_TRACE_ID	Scalar reporting headers, written alongside HEADER_KEY by IsotopeProducerInterceptor.
static final int	MAX_HOPS	Bounded ring for the hop list — a payload guard, not a routing cap.
static final int	TRACE_ID_BYTES

Method Summary

Modifier and Type	Method	Description
Isotope	appendHop(Isotope.Hop hop)
static Isotope	fromHeaders(org.apache.kafka.common.header.Headers headers)	Returns the isotope carried by the given headers, or null if absent.
static Isotope	fromJsonBytes(byte[] bytes)
List<Isotope.Hop>	hops()
static Isotope	newTrace(String originService)	Starts a fresh trace with an unnamed pipeline ("unknown").
static Isotope	newTrace(String originService, String pipeline)
static Isotope	newTrace(String originService, String pipeline, long originTsMs)	Starts a fresh trace whose origin timestamp is set explicitly (e.g.
String	originService()
long	originTsMs()
String	pipeline()
byte[]	toJsonBytes()
byte[]	traceId()
String	traceIdHex()
boolean	truncated()
static String	uuidV7String()	Convenience: a fresh UUIDv7 in canonical xxxxxxxx-...-7xxx-yxxx-... string form.

Methods inherited from class java.lang.Object

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

Field Details

HEADER_KEY

public static final String HEADER_KEY

See Also:

• Constant Field Values

HEADER_TRACE_ID

public static final String HEADER_TRACE_ID

Scalar reporting headers, written alongside HEADER_KEY by IsotopeProducerInterceptor. All values are UTF-8 strings so Flink SQL can cast them directly to STRING (or BIGINT for the numeric ones) without a UDF.

See Also:

• Constant Field Values

HEADER_ORIGIN_TS

public static final String HEADER_ORIGIN_TS

See Also:

• Constant Field Values

HEADER_ORIGIN_SERVICE

public static final String HEADER_ORIGIN_SERVICE

See Also:

• Constant Field Values

HEADER_PIPELINE

public static final String HEADER_PIPELINE

See Also:

• Constant Field Values

HEADER_THIS_SERVICE

public static final String HEADER_THIS_SERVICE

See Also:

• Constant Field Values

HEADER_THIS_TOPIC

public static final String HEADER_THIS_TOPIC

See Also:

• Constant Field Values

HEADER_HOP_COUNT

public static final String HEADER_HOP_COUNT

See Also:

• Constant Field Values

HEADER_CONSUMER_SERVICE

public static final String HEADER_CONSUMER_SERVICE

Consume-side marker header. Written by IsotopeContext.recordConsume when a consumer emits a marker to the isotope_consume_edge_markers topic, on top of the seven forwarded headers above. Its presence is what Flink uses to distinguish a consume-event marker from a regular produced record.

See Also:

• Constant Field Values

MAX_HOPS

public static final int MAX_HOPS

Bounded ring for the hop list — a payload guard, not a routing cap. On overflow the oldest hop is evicted and truncated latches true; the trace keeps flowing through as many services as you want.

See Also:

• Constant Field Values

TRACE_ID_BYTES

public static final int TRACE_ID_BYTES

See Also:

• Constant Field Values

Method Details

newTrace

public static Isotope newTrace(String originService)

Starts a fresh trace with an unnamed pipeline ("unknown").

newTrace

public static Isotope newTrace(String originService,
 String pipeline)

newTrace

public static Isotope newTrace(String originService,
 String pipeline,
 long originTsMs)

Starts a fresh trace whose origin timestamp is set explicitly (e.g. taken from the originating event rather than wall-clock now). The trace id is a UUIDv7 seeded from originTsMs so id order still tracks origin time.

uuidV7String

public static String uuidV7String()

Convenience: a fresh UUIDv7 in canonical xxxxxxxx-...-7xxx-yxxx-... string form.

traceId

public byte[] traceId()

traceIdHex

public String traceIdHex()

originTsMs

public long originTsMs()

originService

public String originService()

pipeline

public String pipeline()

hops

public List<Isotope.Hop> hops()

truncated

public boolean truncated()

appendHop

public Isotope appendHop(Isotope.Hop hop)

toJsonBytes

public byte[] toJsonBytes()

fromJsonBytes

public static Isotope fromJsonBytes(byte[] bytes)

fromHeaders

public static Isotope fromHeaders(org.apache.kafka.common.header.Headers headers)

Returns the isotope carried by the given headers, or null if absent.