kafka

package
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Sep 2, 2026 License: MIT Imports: 15 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.

func WaitModeForAckContract 

func WaitModeForAckContract(contract AckContract) (effectusruntime.WaitMode, error)

WaitModeForAckContract maps the externally documented offset boundary to the exact Engine.Execute wait mode. Callers must reject any other value.

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"`
	StartOffset      string   `json:"start_offset" yaml:"start_offset"`

	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 the daemon Kafka-admission 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 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 admits Kafka records through the daemon's durable execution boundary.

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) 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.

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"
)

type PostgresAttemptTracker 

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

PostgresAttemptTracker records Kafka handler failures in the durable daemon ledger. It is safe to share between consumer instances in one consumer group.

func NewPostgresAttemptTracker 

func NewPostgresAttemptTracker(db *sql.DB) (*PostgresAttemptTracker, error)

NewPostgresAttemptTracker creates the production attempt tracker. The caller must have applied the Kafka delivery ledger migration before consumption.

func (*PostgresAttemptTracker) Attempts 

func (tracker *PostgresAttemptTracker) Attempts(ctx context.Context, deliveryID string) (int, error)

func (*PostgresAttemptTracker) ClearAttempts 

func (tracker *PostgresAttemptTracker) ClearAttempts(ctx context.Context, deliveryID string) error

func (*PostgresAttemptTracker) RecordFailure 

func (tracker *PostgresAttemptTracker) RecordFailure(ctx context.Context, deliveryID string) (int, error)

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.