Class ConfluentWireFormat

java.lang.Object

io.krabka.streams.schema.ConfluentWireFormat

public final class ConfluentWireFormat
extends Object

Encodes and decodes Confluent schema registry frames.

Every framed record starts with a five-byte header: the MAGIC byte followed by the schema ID as a big-endian 32-bit integer. Protobuf records insert a zigzag-varint message-index list between the header and the body; the common case of the first top-level message is encoded as the single byte 0.

The serdes in this package frame records automatically. Use this class directly when interoperating with foreign framed bytes, for example when inspecting records in tooling or writing a custom serde.

Example

byte[] framed = ConfluentWireFormat.encode(42, body);
var frame = ConfluentWireFormat.decode(framed);
assert frame.schemaId() == 42;
assert Arrays.equals(frame.body(), body);

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	ConfluentWireFormat.Frame	A decoded frame.
static final record	ConfluentWireFormat.ProtobufFrame	A decoded Protobuf frame.

Field Summary

Fields

Modifier and Type	Field	Description
static final byte	MAGIC	The first byte of every Confluent-framed record.

Method Summary

Modifier and Type	Method	Description
static ConfluentWireFormat.Frame	decode(byte[] bytes)	Splits framed record bytes into the schema ID and body.
static ConfluentWireFormat.ProtobufFrame	decodeProtobuf(byte[] bytes)	Splits framed Protobuf record bytes into schema ID, message indexes, and body.
static byte[]	encode(int schemaId, byte[] body)	Frames body bytes with the magic byte and schema ID.
static byte[]	encodeProtobuf(int schemaId, List<Integer> messageIndexes, byte[] body)	Frames Protobuf body bytes, including the message-index list.

Methods inherited from class Object

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

Field Details

MAGIC

public static final byte MAGIC

The first byte of every Confluent-framed record.

See Also:

• Constant Field Values

Method Details

encode

public static byte[] encode(int schemaId,
 byte[] body)

Frames body bytes with the magic byte and schema ID.

Parameters:

schemaId - the registry schema ID to record in the header

body - the encoded record body

Returns:

the framed record bytes

decode

public static ConfluentWireFormat.Frame decode(byte[] bytes)

Splits framed record bytes into the schema ID and body.

Parameters:

bytes - the framed record bytes

Returns:

the decoded frame

Throws:

org.apache.kafka.common.errors.SerializationException - if the input is shorter than five bytes or does not start with MAGIC

encodeProtobuf

public static byte[] encodeProtobuf(int schemaId,
 List<Integer> messageIndexes,
 byte[] body)

Frames Protobuf body bytes, including the message-index list.

The indexes identify the message within its .proto file: each entry is the message's position at its nesting level, outermost first. The common [0] (first top-level message) is written in its compact single-byte form.

Parameters:

schemaId - the registry schema ID to record in the header

messageIndexes - the non-empty message-index path, outermost first

body - the serialized Protobuf message

Returns:

the framed record bytes

Throws:

IllegalArgumentException - if messageIndexes is empty

decodeProtobuf

public static ConfluentWireFormat.ProtobufFrame decodeProtobuf(byte[] bytes)

Splits framed Protobuf record bytes into schema ID, message indexes, and body.

Parameters:

bytes - the framed record bytes

Returns:

the decoded Protobuf frame

Throws:

org.apache.kafka.common.errors.SerializationException - if the input is shorter than five bytes, does not start with MAGIC, or carries a malformed message-index list