kafka

package
v0.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 1, 2026 License: MIT Imports: 18 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrPoisonMessage = errors.New("Kafka poison message halted consumption")

Functions

func DeliveryID 

func DeliveryID(clusterNamespace string, message kafka.Message) string

DeliveryID is stable across consumer restarts and group rebalances.

func ValidateConfig 

func ValidateConfig(config *Config) error

ValidateConfig checks source configuration without opening consumer resources.

Types

type AckContract 

type AckContract string

AckContract selects the condition required before an offset commit. 

const (
	AckAfterDurableAcceptance   AckContract = "durable_acceptance"
	AckAfterCompletedProcessing AckContract = "completed_processing"
)

type AttemptTracker 

type AttemptTracker interface {
	Attempts(context.Context, string) (int, error)
	RecordFailure(context.Context, string) (int, error)
	ClearAttempts(context.Context, string) error
}

AttemptTracker durably counts handler failures by stable delivery identity. A crash, rebalance, or commit outage after a successful handler must not consume a poison attempt.

func NewMemoryAttemptTracker 

func NewMemoryAttemptTracker() AttemptTracker

NewMemoryAttemptTracker returns process-local tracking for tests only. Production consumers must use durable tracking.

type CheckedHandlerConfig 

type CheckedHandlerConfig = EngineHandlerConfig

CheckedHandlerConfig and NewCheckedHandler are compatibility facades. They no longer accept an artifact or a Kafka-specific execution request.

type Config 

type Config struct {
	SourceID         string            `json:"source_id" yaml:"source_id"`
	ClusterNamespace string            `json:"cluster_namespace" yaml:"cluster_namespace"`
	Brokers          []string          `json:"brokers" yaml:"brokers"`
	Topic            string            `json:"topic" yaml:"topic"`
	ConsumerGroup    string            `json:"consumer_group" yaml:"consumer_group"`
	SchemaFormat     string            `json:"schema_format" yaml:"schema_format"`
	StartOffset      string            `json:"start_offset" yaml:"start_offset"`
	FactMappings     map[string]string `json:"fact_mappings" yaml:"fact_mappings"`
	Headers          map[string]string `json:"headers" yaml:"headers"`

	AckContract            AckContract     `json:"ack_contract" yaml:"ack_contract"`
	MaxAttempts            int             `json:"max_attempts" yaml:"max_attempts"`
	InitialBackoff         time.Duration   `json:"initial_backoff" yaml:"initial_backoff"`
	MaxBackoff             time.Duration   `json:"max_backoff" yaml:"max_backoff"`
	CommitTimeout          time.Duration   `json:"commit_timeout" yaml:"commit_timeout"`
	PoisonPolicy           PoisonPolicy    `json:"poison_policy" yaml:"poison_policy"`
	DLQTopic               string          `json:"dlq_topic" yaml:"dlq_topic"`
	DLQDeliveryMode        DLQDeliveryMode `json:"dlq_delivery_mode" yaml:"dlq_delivery_mode"`
	MinBytes               int             `json:"min_bytes" yaml:"min_bytes"`
	MaxBytes               int             `json:"max_bytes" yaml:"max_bytes"`
	HeartbeatInterval      time.Duration   `json:"heartbeat_interval" yaml:"heartbeat_interval"`
	SessionTimeout         time.Duration   `json:"session_timeout" yaml:"session_timeout"`
	RebalanceTimeout       time.Duration   `json:"rebalance_timeout" yaml:"rebalance_timeout"`
	JoinGroupBackoff       time.Duration   `json:"join_group_backoff" yaml:"join_group_backoff"`
	PartitionWatchInterval time.Duration   `json:"partition_watch_interval" yaml:"partition_watch_interval"`
}

Config holds Kafka source configuration

type ConsumerStatus 

type ConsumerStatus struct {
	LastFetchedOffset   int64
	LastCommittedOffset int64
	HighWatermark       int64
	Lag                 int64
	CommitHealthy       bool
}

ConsumerStatus is a nonblocking snapshot of the active Kafka boundary.

type DLQDeliveryMode 

type DLQDeliveryMode string

DLQDeliveryMode describes the crash window between DLQ publication and source-offset commit. kafka-go does not combine these operations here. 

const DLQAtLeastOnceNonTransactional DLQDeliveryMode = "at_least_once_non_transactional"

type Delivery 

type Delivery struct {
	ID      string
	Message kafka.Message
	Attempt int
}

Delivery is one immutable Kafka record with a stable admission identity.

type EngineHandlerConfig 

type EngineHandlerConfig struct {
	Ruleset         string
	Version         string
	DefaultTenant   string
	MaxMessageBytes int
	WaitMode        effectusruntime.WaitMode
}

EngineHandlerConfig maps Kafka delivery semantics to Engine.Execute.

type ExecuteEngine 

type ExecuteEngine interface {
	Execute(context.Context, effectusruntime.ExecuteRequest) (effectusruntime.ExecuteResult, error)
}

ExecuteEngine is the shared runtime.Engine surface used by Kafka.

type Factory 

type Factory struct{}

Factory for Kafka sources

func (*Factory) Create 

func (f *Factory) Create(config adapters.SourceConfig) (adapters.FactSource, error)

func (*Factory) GetConfigSchema 

func (f *Factory) GetConfigSchema() adapters.ConfigSchema

func (*Factory) ValidateConfig 

func (f *Factory) ValidateConfig(config adapters.SourceConfig) error

type HandleResult 

type HandleResult struct {
	DurablyAccepted bool
	Completed       bool
}

HandleResult states which acknowledgement boundary the handler reached.

type Handler 

type Handler interface {
	Handle(context.Context, Delivery) (HandleResult, error)
}

Handler blocks until it reaches an acknowledgement boundary or returns an error.

func NewCheckedHandler 

func NewCheckedHandler(config CheckedHandlerConfig, engine ExecuteEngine) (Handler, error)

func NewEngineHandler 

func NewEngineHandler(config EngineHandlerConfig, engine ExecuteEngine) (Handler, error)

NewEngineHandler creates a Kafka Handler that admits every record through runtime.Engine.Execute. No artifact or transport-specific checked request is accepted at this boundary.

type HandlerFunc 

type HandlerFunc func(context.Context, Delivery) (HandleResult, error)

HandlerFunc adapts a function to Handler.

func (HandlerFunc) Handle 

func (function HandlerFunc) Handle(ctx context.Context, delivery Delivery) (HandleResult, error)

type KafkaSource 

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

KafkaSource implements the FactSource interface for Kafka

func NewKafkaSource 

func NewKafkaSource(config *Config) (*KafkaSource, error)

NewKafkaSource creates a new Kafka fact source.

func (*KafkaSource) ConsumerStatus 

func (k *KafkaSource) ConsumerStatus() ConsumerStatus

func (*KafkaSource) GetMetadata 

func (k *KafkaSource) GetMetadata() adapters.SourceMetadata

GetMetadata implements FactSource.GetMetadata

func (*KafkaSource) GetSourceSchema 

func (k *KafkaSource) GetSourceSchema() *adapters.Schema

GetSourceSchema implements FactSource.GetSourceSchema

func (*KafkaSource) HealthCheck 

func (k *KafkaSource) HealthCheck() error

HealthCheck implements FactSource.HealthCheck

func (*KafkaSource) Ready 

func (k *KafkaSource) Ready() error

Ready reports whether the consumer loop is active and has not crossed a fatal commit or poison boundary. It does not perform a blocking broker dial.

func (*KafkaSource) Run 

func (k *KafkaSource) Run(ctx context.Context, handler Handler) error

Run consumes one application-level record at a time. It fetches no later record until the current record reaches the configured acknowledgement boundary.

func (*KafkaSource) SetAttemptTracker 

func (source *KafkaSource) SetAttemptTracker(tracker AttemptTracker) error

SetAttemptTracker configures durable attempt accounting across rebalances. Call this before Run or Start.

func (*KafkaSource) SetDLQPublisher 

func (source *KafkaSource) SetDLQPublisher(publisher MessagePublisher) error

SetDLQPublisher replaces the default Kafka writer for testing or custom publication.

func (*KafkaSource) SetPoisonAcknowledger 

func (source *KafkaSource) SetPoisonAcknowledger(acknowledger PoisonAcknowledger) error

SetPoisonAcknowledger configures the durable poison audit boundary. Call this before Run or Start.

func (*KafkaSource) Start 

func (k *KafkaSource) Start(ctx context.Context) error

Start implements the legacy channel API. A channel send is only a local handoff, so this mode does not provide end-to-end acknowledgement.

func (*KafkaSource) Stop 

func (k *KafkaSource) Stop(ctx context.Context) error

Stop cancels admission and waits for the active handler to return.

func (*KafkaSource) Subscribe 

func (k *KafkaSource) Subscribe(ctx context.Context, factTypes []string) (<-chan *adapters.TypedFact, error)

Subscribe returns the legacy local handoff channel.

type MessageConverter 

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

MessageConverter converts Kafka messages to TypedFacts

func (*MessageConverter) ConvertMessage 

func (c *MessageConverter) ConvertMessage(msg kafka.Message) (*adapters.TypedFact, error)

ConvertMessage converts a Kafka message to a TypedFact

type MessagePublisher 

type MessagePublisher interface {
	WriteMessages(context.Context, ...kafka.Message) error
}

MessagePublisher publishes a DLQ record and returns only after broker acknowledgement.

type PoisonAcknowledger 

type PoisonAcknowledger interface {
	AcknowledgePoison(context.Context, PoisonDisposition) error
}

PoisonAcknowledger durably records an operator-selected poison disposition.

type PoisonDisposition 

type PoisonDisposition struct {
	DeliveryID string
	Policy     PoisonPolicy
	Attempts   int
	Error      string
	Message    kafka.Message
}

PoisonDisposition is the durable audit input for skip and DLQ policies.

type PoisonPolicy 

type PoisonPolicy string

PoisonPolicy controls records that exhaust the configured attempts. 

const (
	PoisonHalt PoisonPolicy = "halt"
	PoisonSkip PoisonPolicy = "skip"
	PoisonDLQ  PoisonPolicy = "dlq"
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.