Kafka

import "github.com/latolukasz/fluxaorm/v2"

registry := fluxaorm.NewRegistry()

// Kafka pool named "events":
registry.RegisterKafka([]string{"localhost:9092"}, "events", &fluxaorm.KafkaPoolOptions{
    ClientID: "my-service",
})

// Pool with multiple brokers:
registry.RegisterKafka([]string{"localhost:9092", "localhost:9093"}, "events", &fluxaorm.KafkaPoolOptions{
    ClientID: "my-service",
})

// Pool with no options (producer-only):
registry.RegisterKafka([]string{"localhost:9092"}, "events", nil)

// Register a single consumer group:
registry.RegisterKafkaConsumerGroup(
    fluxaorm.NewKafkaConsumerGroup("my-group", "events").Topics("orders", "payments"),
)

// Register multiple consumer groups:
registry.RegisterKafkaConsumerGroup(
    fluxaorm.NewKafkaConsumerGroup("order-group", "events").Topics("orders"),
)
registry.RegisterKafkaConsumerGroup(
    fluxaorm.NewKafkaConsumerGroup("payment-group", "events").Topics("payments"),
)

// Consumer group with all options:
registry.RegisterKafkaConsumerGroup(
    fluxaorm.NewKafkaConsumerGroup("my-group", "events").
        Topics("orders", "payments").
        SessionTimeout(30 * time.Second).
        RebalanceTimeout(60 * time.Second).
        FetchMaxBytes(10485760).
        AutoCommitInterval(5 * time.Second),
)

type KafkaPoolOptions struct {
    ClientID              string
    RequiredAcks          int
    ProducerLinger        time.Duration
    MaxBufferedRecords    int
    SASL                  *KafkaSASLConfig
    IgnoredTopics         []string
    IgnoredConsumerGroups []string
}

cg := fluxaorm.NewKafkaConsumerGroup("my-group", "events").
    Topics("orders", "payments").
    SessionTimeout(30 * time.Second).
    FetchMaxBytes(10485760)

type ConfigKafkaConsumerGroup struct {
    Name                 string   `yaml:"name"`
    Topics               []string `yaml:"topics"`
    SessionTimeoutMs     int      `yaml:"sessionTimeoutMs"`
    RebalanceTimeoutMs   int      `yaml:"rebalanceTimeoutMs"`
    FetchMaxBytes        int      `yaml:"fetchMaxBytes"`
    AutoCommitIntervalMs int      `yaml:"autoCommitIntervalMs"`
}

type ConfigKafkaTopic struct {
    Name              string            `yaml:"name"`
    Partitions        int32             `yaml:"partitions"`
    ReplicationFactor int16             `yaml:"replicationFactor"`
    Configs           map[string]string `yaml:"configs"`
}

type ConfigKafka struct {
    Code                  string                     `yaml:"code"`
    Brokers               []string                   `yaml:"brokers"`
    ClientID              string                     `yaml:"clientID"`
    RequiredAcks          int                        `yaml:"requiredAcks"`
    ProducerLingerMs      int                        `yaml:"producerLingerMs"`
    MaxBufferedRecords    int                        `yaml:"maxBufferedRecords"`
    SASLMechanism         string                     `yaml:"saslMechanism"`
    SASLUser              string                     `yaml:"saslUser"`
    SASLPassword          string                     `yaml:"saslPassword"`
    ConsumerGroups        []ConfigKafkaConsumerGroup `yaml:"consumerGroups"`
    IgnoredTopics         []string                   `yaml:"ignoredTopics"`
    IgnoredConsumerGroups []string                   `yaml:"ignoredConsumerGroups"`
    Topics                []ConfigKafkaTopic         `yaml:"topics"`
}

pool := engine.Kafka("events")

cg, err := pool.ConsumerGroup("my-group")
if err != nil {
    // name not registered or connection failed
}
defer cg.Close()

cg := pool.MustConsumerGroup("my-group")
defer cg.Close()

for _, name := range pool.ConsumerGroupNames() {
    fmt.Println(name)
}

type KafkaRecord struct {
    Topic     string
    Key       []byte
    Value     []byte
    Headers   []KafkaRecordHeader
    Partition int32
    Offset    int64
    Timestamp time.Time
}

type KafkaRecordHeader struct {
    Key   string
    Value []byte
}

pool := engine.Kafka("events")

record := &fluxaorm.KafkaRecord{
    Topic: "orders",
    Key:   []byte("order-123"),
    Value: []byte(`{"status":"created"}`),
    Headers: []fluxaorm.KafkaRecordHeader{
        {Key: "source", Value: []byte("api")},
    },
}

err := pool.ProduceSync(ctx, record)
if err != nil {
    // handle error
}

pool := engine.Kafka("events")

record := &fluxaorm.KafkaRecord{
    Topic: "orders",
    Key:   []byte("order-456"),
    Value: []byte(`{"status":"shipped"}`),
}

pool.Produce(ctx, record, func(r *fluxaorm.KafkaRecord, err error) {
    if err != nil {
        log.Printf("produce failed: %v", err)
    }
})

pool := engine.Kafka("events")
cg := pool.MustConsumerGroup("my-group")
defer cg.Close()

fetches := cg.PollFetches(ctx)
if fetches.IsEmpty() {
    return
}

fetches.EachError(func(topic string, partition int32, err error) {
    log.Printf("fetch error topic=%s partition=%d: %v", topic, partition, err)
})

fetches.EachRecord(func(record *fluxaorm.KafkaRecord) {
    fmt.Printf("topic=%s key=%s value=%s\n", record.Topic, record.Key, record.Value)
})

err := cg.CommitUncommittedOffsets(ctx)
if err != nil {
    // handle error
}

cg := pool.MustConsumerGroup("my-group")
defer cg.Close()
client := cg.GetKgoClient()
// use the franz-go client directly for advanced operations

registry.RegisterKafka([]string{"kafka-broker:9093"}, "secure-events", &fluxaorm.KafkaPoolOptions{
    ClientID: "my-service",
    SASL: &fluxaorm.KafkaSASLConfig{
        Mechanism: "SCRAM-SHA-256",
        User:      "kafka-user",
        Password:  "kafka-password",
    },
})
registry.RegisterKafkaConsumerGroup(
    fluxaorm.NewKafkaConsumerGroup("my-group", "secure-events").Topics("orders"),
)

import fluxaorm "github.com/latolukasz/fluxaorm/v2"

topic := fluxaorm.NewKafkaTopic("orders", "default").
    Partitions(6).
    ReplicationFactor(3).
    RetentionMs(86400000).
    CleanupPolicy("delete").
    MinInsyncReplicas(2)

registry := fluxaorm.NewRegistry()
registry.RegisterKafka([]string{"localhost:9092"}, "default", nil)

registry.RegisterKafkaTopic(
    fluxaorm.NewKafkaTopic("orders", "default").
        Partitions(6).
        ReplicationFactor(3).
        RetentionMs(86400000).
        CleanupPolicy("delete").
        MinInsyncReplicas(2),
)

registry.RegisterKafkaTopic(
    fluxaorm.NewKafkaTopic("payments", "default").
        Partitions(3).
        ReplicationFactor(3).
        RetentionBytes(1073741824).
        MaxMessageBytes(1048576),
)

engine, err := registry.Validate()

ctx := engine.NewContext(context.Background())

alters, err := fluxaorm.GetKafkaAlters(ctx)
if err != nil {
    panic(err)
}
for _, alter := range alters {
    fmt.Println(alter.Description) // e.g. "Create topic 'orders' with 6 partitions"
    fmt.Println(alter.Pool)        // e.g. "default"
}

for _, alter := range alters {
    err = alter.Exec(ctx)
    if err != nil {
        panic(err)
    }
}

registry.RegisterKafka([]string{"localhost:9092"}, "default", &fluxaorm.KafkaPoolOptions{
    IgnoredTopics: []string{"legacy-topic", "external-service-topic"},
})

registry.RegisterKafka([]string{"localhost:9092"}, "default", &fluxaorm.KafkaPoolOptions{
    IgnoredConsumerGroups: []string{"legacy-consumer", "external-service-consumer"},
})

config := &fluxaorm.Config{
    KafkaPools: []fluxaorm.ConfigKafka{
        {
            Code:                  "default",
            Brokers:               []string{"localhost:9092"},
            IgnoredTopics:         []string{"legacy-topic"},
            IgnoredConsumerGroups: []string{"legacy-consumer"},
            Topics: []fluxaorm.ConfigKafkaTopic{
                {
                    Name:              "orders",
                    Partitions:        6,
                    ReplicationFactor: 3,
                    Configs: map[string]string{
                        "retention.ms":         "86400000",
                        "cleanup.policy":       "delete",
                        "min.insync.replicas":  "2",
                    },
                },
            },
        },
    },
}

err := registry.InitByConfig(config)

package main

import (
    "context"
    "fmt"

    "github.com/latolukasz/fluxaorm/v2"
)

func main() {
    registry := fluxaorm.NewRegistry()
    registry.RegisterKafka([]string{"localhost:9092"}, "default", &fluxaorm.KafkaPoolOptions{
        IgnoredTopics:         []string{"legacy-topic"},
        IgnoredConsumerGroups: []string{"legacy-consumer"},
    })

    // Define consumer groups
    registry.RegisterKafkaConsumerGroup(
        fluxaorm.NewKafkaConsumerGroup("order-group", "default").Topics("orders"),
    )
    registry.RegisterKafkaConsumerGroup(
        fluxaorm.NewKafkaConsumerGroup("payment-group", "default").Topics("payments"),
    )

    // Define topics
    registry.RegisterKafkaTopic(
        fluxaorm.NewKafkaTopic("orders", "default").
            Partitions(6).
            ReplicationFactor(3).
            RetentionMs(86400000).
            CleanupPolicy("delete").
            MinInsyncReplicas(2),
    )

    registry.RegisterKafkaTopic(
        fluxaorm.NewKafkaTopic("payments", "default").
            Partitions(3).
            ReplicationFactor(3).
            RetentionBytes(1073741824).
            CompressionType("snappy"),
    )

    engine, err := registry.Validate()
    if err != nil {
        panic(err)
    }

    ctx := engine.NewContext(context.Background())

    // Get and apply alters
    alters, err := fluxaorm.GetKafkaAlters(ctx)
    if err != nil {
        panic(err)
    }
    for _, alter := range alters {
        fmt.Println(alter.Description)
        err = alter.Exec(ctx)
        if err != nil {
            panic(err)
        }
    }
}

registry.RegisterKafka([]string{"localhost:9092"}, "events", nil)
registry.RegisterAsyncFlush("events", &fluxaorm.AsyncFlushOptions{
    TopicPartitions: 6,
})

fluxaorm.AsyncSQLTopicName           // "_fluxa_async_sql"
fluxaorm.AsyncSQLDeadLetterTopicName // "_fluxa_async_sql_failed"

cg := pool.MustConsumerGroup("my-group")
defer cg.Close()
// ... consume records ...

pool.Close()