Documentation
¶
Overview ¶
Package flowkafka provides a high-level abstraction for working with Apache Kafka in Go applications.
It wraps segmentio/kafka-go behind clean interfaces so you can produce, consume, and manage Kafka messages with minimal boilerplate.
The package exposes two independent clients:
- [KafkaClient] — produce JSON or Avro messages, consume with manual commit, validate broker connectivity, and manage reader lifecycle.
- [SchemaRegistryClient] — create, query, and validate schemas against any Confluent-compatible Schema Registry.
Both clients are defined as interfaces, making them straightforward to mock in unit tests.
See the project README and the examples/ directory for runnable demos.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Consumer ¶
Consumer is an interface that defines the methods for interacting with a Kafka consumer.
func NewConsumer ¶
func NewConsumer(config ConsumerConfig) Consumer
NewConsumer creates a new Kafka consumer with the given configuration.
type ConsumerConfig ¶
ConsumerConfig holds the configuration options for the Kafka consumer.
type Producer ¶
Producer is an interface that defines the methods for interacting with a Kafka producer.
func NewProducer ¶
func NewProducer(config ProducerConfig) Producer
NewProducer creates a new Kafka producer with the given configuration.
type ProducerConfig ¶
ProducerConfig holds the configuration options for the Kafka producer.
type SchemaRegistry ¶
type SchemaRegistry = schemaregistry.SchemaRegistry
SchemaRegistry defines the interface for interacting with a schema registry service. It provides methods for retrieving schemas, validating connections, and creating new schemas. Implementations of this interface should handle communication with schema registry backends such as Confluent Schema Registry or compatible services.
func NewSchemaRegistry ¶
func NewSchemaRegistry(config SchemaRegistryConfig) SchemaRegistry
NewSchemaRegistry creates a new SchemaRegistry instance using the provided configuration. It initializes the underlying schema registry client and returns a wrapper that implements the SchemaRegistry interface.
type SchemaRegistryConfig ¶
type SchemaRegistryConfig = schemaregistry.Config
SchemaRegistryConfig holds the configuration parameters for connecting to a Kafka Schema Registry. It contains the endpoint URL and SASL credentials required for authentication.
type SchemaType ¶
type SchemaType = schemaregistry.SchemaType
SchemaType is a type alias for srclient.SchemaType that represents the format of a schema in the schema registry (e.g., AVRO, JSON, PROTOBUF).
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
internal
|
|
|
consumer
Package consumer provides a Kafka consumer implementation with consumer group support.
|
Package consumer provides a Kafka consumer implementation with consumer group support. |
|
producer
Package producer provides a Kafka producer implementation that supports retry logic and clean shutdown.
|
Package producer provides a Kafka producer implementation that supports retry logic and clean shutdown. |
|
schemaregistry
Package schemaregistry provides a client for interacting with the Kafka schema registry.
|
Package schemaregistry provides a client for interacting with the Kafka schema registry. |