avro

Documentation

Index

• func NewAvroEncoder(namespace string, schemaM SchemaManager, config *common.Config) codec.RowEventEncoder
• func NewBatchEncoderBuilder(ctx context.Context, config *common.Config) (codec.RowEventEncoderBuilder, error)
• func NewDecoder(config *common.Config, schemaM SchemaManager, topic string, db *sql.DB) codec.RowEventDecoder
• func TeardownEncoderAndSchemaRegistry4Testing()
• type BatchEncoder
  • func SetupEncoderAndSchemaRegistry4Testing(ctx context.Context, config *common.Config) (*BatchEncoder, error)
  • func (a *BatchEncoder) AppendRowChangedEvent(ctx context.Context, topic string, e *model.RowChangedEvent, callback func()) error
  • func (a *BatchEncoder) Build() (messages []*common.Message)
  • func (a *BatchEncoder) EncodeCheckpointEvent(ts uint64) (*common.Message, error)
  • func (a *BatchEncoder) EncodeDDLEvent(e *model.DDLEvent) (*common.Message, error)
• type SchemaGenerator
• type SchemaManager
  • func NewConfluentSchemaManager(ctx context.Context, registryURL string, credential *security.Credential) (SchemaManager, error)
  • func NewGlueSchemaManager(ctx context.Context, cfg *config.GlueSchemaRegistryConfig) (SchemaManager, error)

Functions

func NewAvroEncoder

func NewAvroEncoder(namespace string, schemaM SchemaManager, config *common.Config) codec.RowEventEncoder

NewAvroEncoder return a avro encoder.

func NewBatchEncoderBuilder

func NewBatchEncoderBuilder(
	ctx context.Context, config *common.Config,
) (codec.RowEventEncoderBuilder, error)

NewBatchEncoderBuilder creates an avro batchEncoderBuilder.

func NewDecoder

func NewDecoder(
	config *common.Config,
	schemaM SchemaManager,
	topic string,
	db *sql.DB,
) codec.RowEventDecoder

NewDecoder return an avro decoder

func TeardownEncoderAndSchemaRegistry4Testing

func TeardownEncoderAndSchemaRegistry4Testing()

TeardownEncoderAndSchemaRegistry4Testing stop the local schema registry for testing.

Types

type BatchEncoder

type BatchEncoder struct {
	// contains filtered or unexported fields
}

BatchEncoder converts the events to binary Avro data

func SetupEncoderAndSchemaRegistry4Testing

func SetupEncoderAndSchemaRegistry4Testing(
	ctx context.Context,
	config *common.Config,
) (*BatchEncoder, error)

SetupEncoderAndSchemaRegistry4Testing start a local schema registry for testing.

func (*BatchEncoder) AppendRowChangedEvent

func (a *BatchEncoder) AppendRowChangedEvent(
	ctx context.Context,
	topic string,
	e *model.RowChangedEvent,
	callback func(),
) error

AppendRowChangedEvent appends a row change event to the encoder NOTE: the encoder can only store one RowChangedEvent!

func (*BatchEncoder) Build

func (a *BatchEncoder) Build() (messages []*common.Message)

Build Messages

func (*BatchEncoder) EncodeCheckpointEvent

func (a *BatchEncoder) EncodeCheckpointEvent(ts uint64) (*common.Message, error)

EncodeCheckpointEvent only encode checkpoint event if the watermark event is enabled it's only used for the testing purpose.

func (*BatchEncoder) EncodeDDLEvent

func (a *BatchEncoder) EncodeDDLEvent(e *model.DDLEvent) (*common.Message, error)

EncodeDDLEvent only encode DDL event if the watermark event is enabled it's only used for the testing purpose.

type SchemaGenerator

type SchemaGenerator func() (string, error)

SchemaGenerator represents a function that returns an Avro schema in JSON. Used for lazy evaluation

type SchemaManager

type SchemaManager interface {
	Register(ctx context.Context, schemaName string, schemaDefinition string) (schemaID, error)
	Lookup(ctx context.Context, schemaName string, schemaID schemaID) (*goavro.Codec, error)
	GetCachedOrRegister(ctx context.Context, topicName string,
		tableVersion uint64, schemaGen SchemaGenerator) (*goavro.Codec, []byte, error)
	RegistryType() string
	ClearRegistry(ctx context.Context, schemaName string) error
}

SchemaManager is an interface for schema registry

func NewConfluentSchemaManager

func NewConfluentSchemaManager(
	ctx context.Context,
	registryURL string,
	credential *security.Credential,
) (SchemaManager, error)

NewConfluentSchemaManager create schema managers, and test connectivity to the schema registry

func NewGlueSchemaManager

func NewGlueSchemaManager(
	ctx context.Context,
	cfg *config.GlueSchemaRegistryConfig,
) (SchemaManager, error)

NewGlueSchemaManager creates a new schema manager for AWS Glue Schema Registry It will load the default AWS credentials if no credentials are provided. It will check if the registry exists, if not, it will return an error.