kafkaconnect

Documentation

Overview

Package kafkaconnect contains utility functions and structures for processing Kafka Connect compatible data.

Index

• func SortFields(s *Schema)
• type DebeziumOp
• type DebeziumPayload
  • func (p DebeziumPayload) ToEnvelope() Envelope
• type DebeziumTransaction
• type Envelope
• type Schema
  • func Reflect(v any) *Schema
• type Type

Functions

func SortFields

func SortFields(s *Schema)

SortFields can be used in tests to make sure the order of fields in a map is deterministic.

Types

type DebeziumOp

type DebeziumOp string

const (
	DebeziumOpCreate DebeziumOp = "c"
	DebeziumOpUpdate DebeziumOp = "u"
	DebeziumOpDelete DebeziumOp = "d"
	DebeziumOpRead   DebeziumOp = "r" // snapshot
)

type DebeziumPayload

type DebeziumPayload struct {
	Before map[string]any `json:"before"`
	After  map[string]any `json:"after"`
	Source any            `json:"source"`
	Op     DebeziumOp     `json:"op"`
	// TimestampMillis is what we call ReadAt in Conduit.
	TimestampMillis int64                `json:"ts_ms,omitempty"`
	Transaction     *DebeziumTransaction `json:"transaction"`
}

func (DebeziumPayload) ToEnvelope

func (p DebeziumPayload) ToEnvelope() Envelope

type DebeziumTransaction

type DebeziumTransaction struct {
	ID                  string `json:"id"`
	TotalOrder          int64  `json:"total_order"`
	DataCollectionOrder int64  `json:"data_collection_order"`
}

type Envelope

type Envelope struct {
	Schema  Schema `json:"schema"`
	Payload any    `json:"payload"`
}

Envelope represents a kafka connect message that includes a schema.

type Schema

type Schema struct {
	Type     Type   `json:"type"`
	Optional bool   `json:"optional,omitempty"`
	Name     string `json:"name,omitempty"`
	Version  int    `json:"version,omitempty"`
	Doc      string `json:"doc,omitempty"`
	Default  string `json:"default,omitempty"`

	Parameters map[string]string `json:"parameters,omitempty"`

	// Type: Array
	Items *Schema `json:"items,omitempty"`
	// Type: Map
	Keys   *Schema `json:"keys,omitempty"`
	Values *Schema `json:"values,omitempty"`
	// Type: Struct
	Fields []Schema `json:"fields,omitempty"`
	// Struct fields
	Field string `json:"field,omitempty"`
}

Schema represents a kafka connect JSON schema, the one that can be included in a JSON message directly.

func Reflect

func Reflect(v any) *Schema

Reflect uses reflection to extract a kafka connect compatible schema from v.

type Type

type Type string

const (
	TypeBoolean Type = "boolean"
	TypeInt8    Type = "int8"
	TypeInt16   Type = "int16"
	TypeInt32   Type = "int32"
	TypeInt64   Type = "int64"
	TypeFloat   Type = "float"
	TypeDouble  Type = "double"
	TypeBytes   Type = "bytes"
	TypeString  Type = "string"
	TypeArray   Type = "array"
	TypeMap     Type = "map"
	TypeStruct  Type = "struct"
)