Overview

The project is highly inspired by Kafka Connect, but implemented with Go. The project contains several connector classes designed for Change-Data-Capture (CDC) pipeline from specific source to sink. Some of them use internal storage based on Badger embedded database to keep offsets/log positions/other important metadata to prevent from data loss or duplication on connection issues. Internal storage is organized in "data" folder in a workspace of running app.

Keep in my mind to not mix connectors in common workspace unless they have unique names ("name" parameter in YAML configuration). It's not recommended to recreate Kafka topics and use the same state of connector's internal storage. If you have to then you might want to remove "data" folder also, but...

Installation

Docker image:

docker pull egsam98/kafka-pipe:{tag}

https://hub.docker.com/r/egsam98/kafka-pipe/tags

Go module:

go get github.com/egsam98/kafka-pipe

Supported connector classes:

Get started

To start application you want to pass YAML configuration as an argument for application.

./{kafka-pipe bin} config.yaml

Common parameters of config:

name: "name of your connector"
class: "any class described in \"Supported classes\" section"
log:
  pretty: true
  level: "info"

Additional params are particular for every class.

pg.Source

Transmits events from PostgreSQL to Kafka topics via logical replication and "pgoutput" plugin. Postgres' last read LSN is stored on disk of internal storage. Config parameters

pg.Snapshot

Selects all/certain rows from tables and supplies to Kafka topics. Connector stops when all rows are produced

Config parameters

ch.Sink

Transmits events from Kafka topics to ClickHouse. Deduplication mechanism is designed similarly to https://github.com/ClickHouse/clickhouse-kafka-connect/blob/main/docs/DESIGN.md#state-machine . Kafka partition offsets are additionally stored on disk of internal storage. Config parameters

routes:
    public.transaction_\d+: transactions

Non-YAML configuration (available if you use connector as go-module):

func BeforeInsert(ctx context.Context, serde Serde, batch []*kgo.Record) ([]any, error)

Hook function that's called per every consumed batch before inserting one into ClickHouse. Due to specific ClickHouse implementations (see here and here) result must consist of struct pointers, i.e.

func BeforeInsert(ctx context.Context, serde Serde, batch []*kgo.Record) ([]any, error) {
	// Handle batch
	// ...
	return []any{&Data{Column1: 1}, &Data{Column1: 2}}, nil
}

s3.Sink

Transmits events from Kafka topics to S3-compatible storage. This connector polls Kafka records according batch settings (see kafka:batch:*), groups by truncated datetime + partition and then uploads collected data chunks into S3. The storage keys are built by following schema: {topic}/{truncated datetime}/{partition}/{min_offset in chunk}-{max_offset in chunk}.gz. If previous data chunk doesn't exceed 5MB limit it's merged with new chunk and deleted. If prev. chunk's max offset >= new chunk's max offset, new chunk is ignored.

Config parameters

s3.Backup

Restores Kafka messages from S3 storage back to topics. The connector stops when all selected objects are handled. Read offsets are saved to internal storage and skipped on rescan afterwards.

Config parameters:

SASL

Specific settings for every SASL protocol. All params are optional.

plain

scram-256

scram-512

See scram-256

oath

aws

Serde

i.e. Serialization and Deserialization. Supported formats: json, avro. Depending on the selected format, different settings are provided:

json

avro

Examples of Serde configuration:

serde:
  format: json
  time_format: rfc3339

serde:
  format: avro
  schemas:
    topic_test: file://schema.avsc