优化结构

2026-08-30 01:08:10 +08:00
parent 509df03d61
commit e0c35e4e9b
23 changed files with 1513 additions and 38 deletions
@@ -0,0 +1,592 @@
+package mq
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	kafkago "github.com/segmentio/kafka-go"
+	"github.com/segmentio/kafka-go/sasl"
+	"github.com/segmentio/kafka-go/sasl/plain"
+)
+
+type KafkaConfig struct {
+	Enabled                bool
+	Brokers                []string
+	ClientID               string
+	GroupID                string
+	Username               string
+	Password               string
+	TLS                    bool
+	TLSSkipVerify          bool
+	StartOffset            string
+	MinBytes               int
+	MaxBytes               int
+	MaxWait                time.Duration
+	ConnectTimeout         time.Duration
+	ReconnectInterval      time.Duration
+	AllowAutoTopicCreation bool
+}
+
+type kafkaSubscription struct {
+	reader  *kafkago.Reader
+	cancel  context.CancelFunc
+	qos     byte
+	handler Handler
+}
+
+type kafkaTopicMetadataError struct {
+	err error
+}
+
+func (e *kafkaTopicMetadataError) Error() string { return e.err.Error() }
+func (e *kafkaTopicMetadataError) Unwrap() error { return e.err }
+
+// Kafka adapts Kafka topics and consumer groups to the shared messaging
+// contract. QoS 0 commits before dispatch; QoS 1 commits after dispatch.
+type Kafka struct {
+	opMu          sync.Mutex
+	mu            sync.RWMutex
+	config        KafkaConfig
+	dialer        *kafkago.Dialer
+	metadata      *kafkago.Client
+	transport     *kafkago.Transport
+	bestEffort    *kafkago.Writer
+	acknowledged  *kafkago.Writer
+	subscriptions map[string]*kafkaSubscription
+	connected     atomic.Bool
+	reconnectAt   atomic.Int64
+	closed        bool
+}
+
+func NewKafka(config KafkaConfig) (*Kafka, error) {
+	client := &Kafka{subscriptions: make(map[string]*kafkaSubscription)}
+	if !config.Enabled {
+		return client, nil
+	}
+	config = defaultKafkaConfig(config)
+	if err := validateKafkaConfig(config); err != nil {
+		return nil, err
+	}
+
+	tlsConfig, mechanism := kafkaSecurity(config)
+	dialer := &kafkago.Dialer{
+		ClientID:      config.ClientID,
+		Timeout:       config.ConnectTimeout,
+		TLS:           tlsConfig,
+		SASLMechanism: mechanism,
+	}
+	transport := &kafkago.Transport{
+		DialTimeout: config.ConnectTimeout,
+		ClientID:    config.ClientID,
+		TLS:         tlsConfig,
+		SASL:        mechanism,
+	}
+	metadata := &kafkago.Client{
+		Addr:      kafkago.TCP(config.Brokers...),
+		Timeout:   config.ConnectTimeout,
+		Transport: transport,
+	}
+	if err := probeKafkaMetadata(context.Background(), metadata, ""); err != nil {
+		transport.CloseIdleConnections()
+		return nil, fmt.Errorf("connect kafka: %w", err)
+	}
+	newWriter := func(acks kafkago.RequiredAcks) *kafkago.Writer {
+		return &kafkago.Writer{
+			Addr:                   kafkago.TCP(config.Brokers...),
+			Balancer:               &kafkago.LeastBytes{},
+			RequiredAcks:           acks,
+			ReadTimeout:            config.ConnectTimeout,
+			WriteTimeout:           config.ConnectTimeout,
+			Transport:              transport,
+			AllowAutoTopicCreation: config.AllowAutoTopicCreation,
+		}
+	}
+	client.config = config
+	client.dialer = dialer
+	client.metadata = metadata
+	client.transport = transport
+	client.bestEffort = newWriter(kafkago.RequireNone)
+	client.acknowledged = newWriter(kafkago.RequireAll)
+	client.markConnected()
+	return client, nil
+}
+
+func probeKafkaMetadata(ctx context.Context, client *kafkago.Client, topic string) error {
+	if client == nil {
+		return ErrUnavailable
+	}
+	request := &kafkago.MetadataRequest{Topics: []string{}}
+	if topic != "" {
+		request.Topics = []string{topic}
+	}
+	response, err := client.Metadata(nonNilContext(ctx), request)
+	if err != nil {
+		return err
+	}
+	if response == nil || len(response.Brokers) == 0 {
+		return errors.New("kafka metadata contains no brokers")
+	}
+	if topic == "" {
+		return nil
+	}
+	for _, item := range response.Topics {
+		if item.Name != topic {
+			continue
+		}
+		if item.Error != nil {
+			return &kafkaTopicMetadataError{err: item.Error}
+		}
+		if len(item.Partitions) == 0 {
+			return &kafkaTopicMetadataError{err: errors.New("topic has no partitions")}
+		}
+		return nil
+	}
+	return &kafkaTopicMetadataError{err: errors.New("topic metadata was not returned")}
+}
+
+func defaultKafkaConfig(config KafkaConfig) KafkaConfig {
+	brokers := make([]string, 0, len(config.Brokers))
+	for _, broker := range config.Brokers {
+		if broker = strings.TrimSpace(broker); broker != "" {
+			brokers = append(brokers, broker)
+		}
+	}
+	config.Brokers = brokers
+	config.ClientID = strings.TrimSpace(config.ClientID)
+	config.GroupID = strings.TrimSpace(config.GroupID)
+	config.StartOffset = strings.ToLower(strings.TrimSpace(config.StartOffset))
+	if config.StartOffset == "" {
+		config.StartOffset = "earliest"
+	}
+	if config.MinBytes <= 0 {
+		config.MinBytes = 1
+	}
+	if config.MaxBytes <= 0 {
+		config.MaxBytes = 10 << 20
+	}
+	if config.MaxWait <= 0 {
+		config.MaxWait = time.Second
+	}
+	if config.ConnectTimeout <= 0 {
+		config.ConnectTimeout = 10 * time.Second
+	}
+	if config.ReconnectInterval <= 0 {
+		config.ReconnectInterval = 5 * time.Second
+	}
+	return config
+}
+
+func validateKafkaConfig(config KafkaConfig) error {
+	if len(config.Brokers) == 0 {
+		return errors.New("kafka brokers are empty")
+	}
+	for _, broker := range config.Brokers {
+		host, portText, err := net.SplitHostPort(broker)
+		port, parseErr := strconv.Atoi(portText)
+		if err != nil || parseErr != nil || strings.TrimSpace(host) == "" || port < 1 || port > 65535 {
+			return fmt.Errorf("invalid kafka broker %q", broker)
+		}
+	}
+	if config.ClientID == "" {
+		return errors.New("kafka client id is empty")
+	}
+	if config.GroupID == "" {
+		return errors.New("kafka group id is empty")
+	}
+	if (config.Username == "") != (config.Password == "") {
+		return errors.New("kafka username and password must be configured together")
+	}
+	if config.TLSSkipVerify && !config.TLS {
+		return errors.New("kafka tls skip verify requires tls")
+	}
+	if config.StartOffset != "earliest" && config.StartOffset != "latest" {
+		return fmt.Errorf("invalid kafka start offset %q", config.StartOffset)
+	}
+	if config.MinBytes < 1 || config.MaxBytes < config.MinBytes {
+		return errors.New("kafka byte limits are invalid")
+	}
+	return nil
+}
+
+func kafkaSecurity(config KafkaConfig) (*tls.Config, sasl.Mechanism) {
+	var tlsConfig *tls.Config
+	if config.TLS {
+		tlsConfig = &tls.Config{MinVersion: tls.VersionTLS12, InsecureSkipVerify: config.TLSSkipVerify} //nolint:gosec // explicit operator setting
+	}
+	var mechanism sasl.Mechanism
+	if config.Username != "" {
+		mechanism = plain.Mechanism{Username: config.Username, Password: config.Password}
+	}
+	return tlsConfig, mechanism
+}
+
+func (c *Kafka) Publish(ctx context.Context, topic string, payload []byte, qos byte, retain bool) error {
+	topic = strings.TrimSpace(topic)
+	if topic == "" {
+		return errors.New("kafka topic is empty")
+	}
+	if retain {
+		return errors.New("kafka does not support retained messages")
+	}
+	if qos > AtLeastOnce {
+		return fmt.Errorf("kafka supports qos 0 or 1, got %d", qos)
+	}
+	if c == nil {
+		return ErrUnavailable
+	}
+	ctx = nonNilContext(ctx)
+	c.mu.RLock()
+	if c.closed {
+		c.mu.RUnlock()
+		return ErrUnavailable
+	}
+	writer := c.bestEffort
+	if qos == AtLeastOnce {
+		writer = c.acknowledged
+	}
+	config := c.config
+	c.mu.RUnlock()
+	if writer == nil {
+		return ErrUnavailable
+	}
+	err := writeKafkaMessage(ctx, writer, kafkago.Message{Topic: topic, Value: append([]byte(nil), payload...)}, config)
+	if err == nil {
+		c.markConnected()
+	}
+	return err
+}
+
+func writeKafkaMessage(ctx context.Context, writer *kafkago.Writer, message kafkago.Message, config KafkaConfig) error {
+	if writer == nil {
+		return ErrUnavailable
+	}
+	ctx = nonNilContext(ctx)
+	if !config.AllowAutoTopicCreation {
+		return writer.WriteMessages(ctx, message)
+	}
+	attemptCtx, cancel := context.WithTimeout(ctx, config.ConnectTimeout)
+	defer cancel()
+	delay := 100 * time.Millisecond
+	var lastErr error
+	for {
+		lastErr = writer.WriteMessages(attemptCtx, message)
+		if lastErr == nil || !kafkaTopicNotReady(lastErr) {
+			return lastErr
+		}
+		timer := time.NewTimer(delay)
+		select {
+		case <-attemptCtx.Done():
+			timer.Stop()
+			if err := ctx.Err(); err != nil {
+				return err
+			}
+			return lastErr
+		case <-timer.C:
+		}
+		if delay < time.Second {
+			delay *= 2
+			if delay > time.Second {
+				delay = time.Second
+			}
+		}
+	}
+}
+
+func kafkaTopicNotReady(err error) bool {
+	if err == nil {
+		return false
+	}
+	if writeErrors, ok := err.(kafkago.WriteErrors); ok {
+		found := false
+		for _, item := range writeErrors {
+			if item == nil {
+				continue
+			}
+			found = true
+			if !kafkaTopicNotReady(item) {
+				return false
+			}
+		}
+		return found
+	}
+	var kafkaErr kafkago.Error
+	if !errors.As(err, &kafkaErr) {
+		return false
+	}
+	return kafkaErr == kafkago.UnknownTopicOrPartition || kafkaErr == kafkago.LeaderNotAvailable || kafkaErr == kafkago.NotLeaderForPartition
+}
+
+func (c *Kafka) Subscribe(ctx context.Context, topic string, qos byte, handler Handler) error {
+	topic = strings.TrimSpace(topic)
+	if topic == "" {
+		return errors.New("kafka topic is empty")
+	}
+	if qos > AtLeastOnce {
+		return fmt.Errorf("kafka supports qos 0 or 1, got %d", qos)
+	}
+	if handler == nil {
+		return errors.New("kafka handler is nil")
+	}
+	if c == nil {
+		return ErrUnavailable
+	}
+	ctx = nonNilContext(ctx)
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+
+	c.opMu.Lock()
+	defer c.opMu.Unlock()
+	c.mu.RLock()
+	if c.closed || c.dialer == nil || c.metadata == nil {
+		c.mu.RUnlock()
+		return ErrUnavailable
+	}
+	config := c.config
+	dialer := c.dialer
+	metadata := c.metadata
+	c.mu.RUnlock()
+	if err := probeKafkaMetadata(ctx, metadata, topic); err != nil {
+		var topicErr *kafkaTopicMetadataError
+		if errors.As(err, &topicErr) {
+			c.markConnected()
+		} else if ctx.Err() == nil {
+			c.markUnavailable()
+		}
+		return fmt.Errorf("lookup kafka topic %q: %w", topic, err)
+	}
+	if old := c.removeSubscription(topic); old != nil {
+		closeKafkaSubscription(old)
+	}
+	reader := kafkago.NewReader(kafkago.ReaderConfig{
+		Brokers:          append([]string(nil), config.Brokers...),
+		GroupID:          config.GroupID,
+		Topic:            topic,
+		Dialer:           dialer,
+		MinBytes:         config.MinBytes,
+		MaxBytes:         config.MaxBytes,
+		MaxWait:          config.MaxWait,
+		JoinGroupBackoff: config.ReconnectInterval,
+		ReadLagInterval:  -1,
+		StartOffset:      kafkaStartOffset(config.StartOffset),
+	})
+	consumeCtx, cancel := context.WithCancel(context.Background())
+	subscription := &kafkaSubscription{reader: reader, cancel: cancel, qos: qos, handler: handler}
+	c.mu.Lock()
+	if c.closed {
+		c.mu.Unlock()
+		cancel()
+		_ = reader.Close()
+		return ErrUnavailable
+	}
+	c.subscriptions[topic] = subscription
+	c.mu.Unlock()
+	c.markConnected()
+	go c.consume(consumeCtx, subscription)
+	return nil
+}
+
+func kafkaStartOffset(value string) int64 {
+	if value == "latest" {
+		return kafkago.LastOffset
+	}
+	return kafkago.FirstOffset
+}
+
+func (c *Kafka) consume(ctx context.Context, subscription *kafkaSubscription) {
+	for {
+		var message kafkago.Message
+		var err error
+		if subscription.qos == AtMostOnce {
+			message, err = subscription.reader.ReadMessage(ctx)
+		} else {
+			message, err = subscription.reader.FetchMessage(ctx)
+		}
+		if err != nil {
+			if ctx.Err() != nil || errors.Is(err, io.EOF) {
+				return
+			}
+			c.markUnavailable()
+			if !sleepContext(ctx, c.config.ReconnectInterval) {
+				return
+			}
+			continue
+		}
+		c.markConnected()
+		subscription.handler(ctx, Message{Topic: message.Topic, Payload: append([]byte(nil), message.Value...), QoS: subscription.qos})
+		if ctx.Err() != nil {
+			return
+		}
+		if subscription.qos == AtLeastOnce {
+			if err = subscription.reader.CommitMessages(ctx, message); err != nil {
+				c.markUnavailable()
+			}
+		}
+	}
+}
+
+func sleepContext(ctx context.Context, duration time.Duration) bool {
+	timer := time.NewTimer(duration)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return false
+	case <-timer.C:
+		return true
+	}
+}
+
+func (c *Kafka) Unsubscribe(ctx context.Context, topics ...string) error {
+	if len(topics) == 0 {
+		return errors.New("kafka topics are empty")
+	}
+	if c == nil {
+		return ErrUnavailable
+	}
+	ctx = nonNilContext(ctx)
+	normalized := make([]string, len(topics))
+	for index, topic := range topics {
+		normalized[index] = strings.TrimSpace(topic)
+		if normalized[index] == "" {
+			return errors.New("kafka topic is empty")
+		}
+	}
+	c.opMu.Lock()
+	defer c.opMu.Unlock()
+	for _, topic := range normalized {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+		if subscription := c.removeSubscription(topic); subscription != nil {
+			closeKafkaSubscription(subscription)
+		}
+	}
+	return nil
+}
+
+func (c *Kafka) removeSubscription(topic string) *kafkaSubscription {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	subscription := c.subscriptions[topic]
+	delete(c.subscriptions, topic)
+	return subscription
+}
+
+func closeKafkaSubscription(subscription *kafkaSubscription) {
+	if subscription == nil {
+		return
+	}
+	if subscription.cancel != nil {
+		subscription.cancel()
+	}
+	if subscription.reader != nil {
+		_ = subscription.reader.Close()
+	}
+}
+
+func (c *Kafka) Connected() bool {
+	if c == nil {
+		return false
+	}
+	c.mu.RLock()
+	closed := c.closed
+	c.mu.RUnlock()
+	return !closed && c.connected.Load()
+}
+
+// Reconnecting lets the outer reloadable honor the configured retry delay
+// before replacing a client after a runtime broker failure.
+func (c *Kafka) Reconnecting() bool {
+	if c == nil {
+		return false
+	}
+	c.mu.RLock()
+	closed := c.closed
+	c.mu.RUnlock()
+	return !closed && time.Now().UnixNano() < c.reconnectAt.Load()
+}
+
+func (c *Kafka) markConnected() {
+	if c == nil {
+		return
+	}
+	c.mu.RLock()
+	closed := c.closed
+	c.mu.RUnlock()
+	if closed {
+		return
+	}
+	c.reconnectAt.Store(0)
+	c.connected.Store(true)
+}
+
+func (c *Kafka) markUnavailable() {
+	if c == nil {
+		return
+	}
+	c.mu.RLock()
+	closed := c.closed
+	c.mu.RUnlock()
+	if closed {
+		return
+	}
+	c.connected.Store(false)
+	c.reconnectAt.Store(time.Now().Add(c.config.ReconnectInterval).UnixNano())
+}
+
+func (c *Kafka) Close() error {
+	if c == nil {
+		return nil
+	}
+	c.opMu.Lock()
+	defer c.opMu.Unlock()
+	c.mu.Lock()
+	if c.closed {
+		c.mu.Unlock()
+		return nil
+	}
+	c.closed = true
+	c.connected.Store(false)
+	c.reconnectAt.Store(0)
+	subscriptions := make([]*kafkaSubscription, 0, len(c.subscriptions))
+	for topic, subscription := range c.subscriptions {
+		subscriptions = append(subscriptions, subscription)
+		delete(c.subscriptions, topic)
+	}
+	bestEffort := c.bestEffort
+	acknowledged := c.acknowledged
+	transport := c.transport
+	c.bestEffort = nil
+	c.acknowledged = nil
+	c.transport = nil
+	c.dialer = nil
+	c.metadata = nil
+	c.mu.Unlock()
+	for _, subscription := range subscriptions {
+		closeKafkaSubscription(subscription)
+	}
+	var result error
+	if bestEffort != nil {
+		result = errors.Join(result, bestEffort.Close())
+	}
+	if acknowledged != nil {
+		result = errors.Join(result, acknowledged.Close())
+	}
+	if transport != nil {
+		transport.CloseIdleConnections()
+	}
+	return result
+}
@@ -0,0 +1,116 @@
+package mq
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestKafkaRoundTripIntegration(t *testing.T) {
+	rawBrokers := strings.TrimSpace(os.Getenv("KRA_KAFKA_TEST_BROKERS"))
+	if rawBrokers == "" {
+		t.Skip("KRA_KAFKA_TEST_BROKERS is not configured")
+	}
+	brokers := []string{"127.0.0.1:1"}
+	for _, broker := range strings.Split(rawBrokers, ",") {
+		if broker = strings.TrimSpace(broker); broker != "" {
+			brokers = append(brokers, broker)
+		}
+	}
+	if len(brokers) == 1 {
+		t.Fatal("KRA_KAFKA_TEST_BROKERS contains no broker addresses")
+	}
+
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	topics := []string{"kra-review-qos1-" + suffix, "kra-review-qos0-" + suffix}
+	groupID := "kra-review-" + suffix
+	newClient := func(clientID string) *Kafka {
+		client, err := NewKafka(KafkaConfig{
+			Enabled:                true,
+			Brokers:                brokers,
+			ClientID:               clientID,
+			GroupID:                groupID,
+			StartOffset:            "earliest",
+			MinBytes:               1,
+			MaxBytes:               1 << 20,
+			MaxWait:                100 * time.Millisecond,
+			ConnectTimeout:         5 * time.Second,
+			ReconnectInterval:      500 * time.Millisecond,
+			AllowAutoTopicCreation: true,
+		})
+		if err != nil {
+			t.Fatalf("NewKafka(%s): %v", clientID, err)
+		}
+		return client
+	}
+
+	client := newClient("kra-review-producer-" + suffix)
+	t.Cleanup(func() { _ = client.Close() })
+	firstPayloads := [][]byte{[]byte("qos1-first"), []byte("qos0-first")}
+	qosValues := []byte{AtLeastOnce, AtMostOnce}
+	for index, topic := range topics {
+		if err := client.Publish(context.Background(), topic, firstPayloads[index], qosValues[index], false); err != nil {
+			t.Fatalf("publish first message to %s: %v", topic, err)
+		}
+	}
+
+	firstReceived := []chan Message{make(chan Message, 1), make(chan Message, 1)}
+	for index, topic := range topics {
+		index := index
+		if err := client.Subscribe(context.Background(), topic, qosValues[index], func(_ context.Context, message Message) {
+			firstReceived[index] <- message
+		}); err != nil {
+			t.Fatalf("subscribe to %s: %v", topic, err)
+		}
+	}
+	for index, received := range firstReceived {
+		message := waitKafkaMessage(t, received, 30*time.Second)
+		if string(message.Payload) != string(firstPayloads[index]) || message.QoS != qosValues[index] {
+			t.Fatalf("first message[%d] = payload %q qos %d", index, message.Payload, message.QoS)
+		}
+	}
+	// QoS 1 commits after the handler returns. Give the synchronous commit time
+	// to finish before closing the readers and rejoining the same group.
+	time.Sleep(500 * time.Millisecond)
+	if err := client.Unsubscribe(context.Background(), topics...); err != nil {
+		t.Fatal(err)
+	}
+
+	replayClient := newClient("kra-review-replay-" + suffix)
+	t.Cleanup(func() { _ = replayClient.Close() })
+	replayed := []chan Message{make(chan Message, 2), make(chan Message, 2)}
+	for index, topic := range topics {
+		index := index
+		if err := replayClient.Subscribe(context.Background(), topic, qosValues[index], func(_ context.Context, message Message) {
+			replayed[index] <- message
+		}); err != nil {
+			t.Fatalf("replay subscribe to %s: %v", topic, err)
+		}
+	}
+	secondPayloads := [][]byte{[]byte("qos1-second"), []byte("qos0-second")}
+	for index, topic := range topics {
+		if err := replayClient.Publish(context.Background(), topic, secondPayloads[index], qosValues[index], false); err != nil {
+			t.Fatalf("publish second message to %s: %v", topic, err)
+		}
+	}
+	for index, received := range replayed {
+		message := waitKafkaMessage(t, received, 30*time.Second)
+		if string(message.Payload) != string(secondPayloads[index]) {
+			t.Fatalf("message[%d] replayed uncommitted payload %q, want %q", index, message.Payload, secondPayloads[index])
+		}
+	}
+}
+
+func waitKafkaMessage(t *testing.T, messages <-chan Message, timeout time.Duration) Message {
+	t.Helper()
+	select {
+	case message := <-messages:
+		return message
+	case <-time.After(timeout):
+		t.Fatal("timed out waiting for kafka message")
+		return Message{}
+	}
+}
@@ -0,0 +1,234 @@
+package mq
+
+import (
+	"context"
+	"errors"
+	"net"
+	"testing"
+	"time"
+
+	kafkago "github.com/segmentio/kafka-go"
+	metadataapi "github.com/segmentio/kafka-go/protocol/metadata"
+)
+
+type kafkaRoundTripperFunc func(context.Context, net.Addr, kafkago.Request) (kafkago.Response, error)
+
+func (f kafkaRoundTripperFunc) RoundTrip(ctx context.Context, addr net.Addr, request kafkago.Request) (kafkago.Response, error) {
+	return f(ctx, addr, request)
+}
+
+func TestDisabledKafkaIsSafeAndUnavailable(t *testing.T) {
+	client, err := NewKafka(KafkaConfig{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if client.Connected() {
+		t.Fatal("disabled kafka reported connected")
+	}
+	if err = client.Publish(context.Background(), "orders", []byte("test"), AtLeastOnce, false); !errors.Is(err, ErrUnavailable) {
+		t.Fatalf("publish error = %v", err)
+	}
+	if err = client.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestEnabledKafkaRequiresConnectionSettings(t *testing.T) {
+	if _, err := NewKafka(KafkaConfig{Enabled: true}); err == nil {
+		t.Fatal("enabled kafka without brokers should fail")
+	}
+	if _, err := NewKafka(KafkaConfig{Enabled: true, Brokers: []string{"localhost:9092"}}); err == nil {
+		t.Fatal("enabled kafka without client and group ids should fail")
+	}
+}
+
+func TestKafkaConfigRejectsInvalidBrokerAndTLSSettings(t *testing.T) {
+	config := KafkaConfig{Enabled: true, Brokers: []string{"missing-port"}, ClientID: "kra", GroupID: "kra"}
+	if _, err := NewKafka(config); err == nil {
+		t.Fatal("invalid kafka broker was accepted")
+	}
+	config.Brokers = []string{"localhost:9092"}
+	config.TLSSkipVerify = true
+	if _, err := NewKafka(config); err == nil {
+		t.Fatal("tls skip verify without tls was accepted")
+	}
+}
+
+func TestKafkaRejectsUnsupportedMessageSemantics(t *testing.T) {
+	client, err := NewKafka(KafkaConfig{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = client.Publish(context.Background(), "orders", nil, AtMostOnce, true); err == nil {
+		t.Fatal("retained kafka message was accepted")
+	}
+	if err = client.Publish(context.Background(), "orders", nil, ExactlyOnce, false); err == nil {
+		t.Fatal("qos 2 kafka message was accepted")
+	}
+	if err = client.Subscribe(context.Background(), "orders", ExactlyOnce, func(context.Context, Message) {}); err == nil {
+		t.Fatal("qos 2 kafka subscription was accepted")
+	}
+}
+
+func TestKafkaTopicNotReadyClassification(t *testing.T) {
+	for _, err := range []error{
+		kafkago.UnknownTopicOrPartition,
+		kafkago.LeaderNotAvailable,
+		kafkago.NotLeaderForPartition,
+		kafkago.WriteErrors{kafkago.UnknownTopicOrPartition},
+	} {
+		if !kafkaTopicNotReady(err) {
+			t.Fatalf("error %v was not classified as topic-not-ready", err)
+		}
+	}
+	for _, err := range []error{
+		kafkago.TopicAuthorizationFailed,
+		errors.New("network failed"),
+		kafkago.WriteErrors{kafkago.UnknownTopicOrPartition, kafkago.TopicAuthorizationFailed},
+	} {
+		if kafkaTopicNotReady(err) {
+			t.Fatalf("error %v was classified as topic-not-ready", err)
+		}
+	}
+}
+
+func TestKafkaStartOffset(t *testing.T) {
+	if kafkaStartOffset("latest") != -1 {
+		t.Fatal("latest offset was not mapped to kafka last offset")
+	}
+	if kafkaStartOffset("earliest") != -2 {
+		t.Fatal("earliest offset was not mapped to kafka first offset")
+	}
+}
+
+func TestNilKafkaClientIsSafe(t *testing.T) {
+	var client *Kafka
+	if err := client.Publish(context.Background(), "orders", nil, AtMostOnce, false); !errors.Is(err, ErrUnavailable) {
+		t.Fatalf("publish error = %v", err)
+	}
+	if err := client.Subscribe(context.Background(), "orders", AtMostOnce, func(context.Context, Message) {}); !errors.Is(err, ErrUnavailable) {
+		t.Fatalf("subscribe error = %v", err)
+	}
+	if err := client.Unsubscribe(context.Background(), "orders"); !errors.Is(err, ErrUnavailable) {
+		t.Fatalf("unsubscribe error = %v", err)
+	}
+	if err := client.Close(); err != nil {
+		t.Fatalf("close error = %v", err)
+	}
+}
+
+func TestKafkaReconnectDelayExpires(t *testing.T) {
+	client := &Kafka{config: KafkaConfig{ReconnectInterval: 20 * time.Millisecond}}
+	client.markUnavailable()
+	if client.Connected() || !client.Reconnecting() {
+		t.Fatalf("failure state connected=%t reconnecting=%t", client.Connected(), client.Reconnecting())
+	}
+	client.reconnectAt.Store(time.Now().Add(-time.Second).UnixNano())
+	if client.Reconnecting() {
+		t.Fatal("reconnect delay did not expire")
+	}
+	client.markConnected()
+	if !client.Connected() || client.Reconnecting() {
+		t.Fatalf("recovered state connected=%t reconnecting=%t", client.Connected(), client.Reconnecting())
+	}
+}
+
+func TestKafkaUnsubscribeDoesNotWaitForConsumerHandler(t *testing.T) {
+	consumeCtx, cancel := context.WithCancel(context.Background())
+	reader := kafkago.NewReader(kafkago.ReaderConfig{Brokers: []string{"127.0.0.1:9092"}, Topic: "orders", Partition: 0})
+	client := &Kafka{subscriptions: map[string]*kafkaSubscription{
+		"orders": {reader: reader, cancel: cancel},
+	}}
+	done := make(chan error, 1)
+	go func() { done <- client.Unsubscribe(context.Background(), "orders") }()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("unsubscribe waited for the consumer handler")
+	}
+	if consumeCtx.Err() == nil {
+		t.Fatal("consumer context was not canceled")
+	}
+}
+
+func TestKafkaUnsubscribeRejectsBlankTopics(t *testing.T) {
+	client := &Kafka{subscriptions: make(map[string]*kafkaSubscription)}
+	if err := client.Unsubscribe(context.Background(), "  "); err == nil {
+		t.Fatal("blank kafka topic was accepted")
+	}
+}
+
+func TestProbeKafkaMetadataDoesNotAutoCreateTopic(t *testing.T) {
+	var captured *metadataapi.Request
+	client := &kafkago.Client{
+		Addr: kafkago.TCP("kafka-1:9092", "kafka-2:9092"),
+		Transport: kafkaRoundTripperFunc(func(_ context.Context, _ net.Addr, request kafkago.Request) (kafkago.Response, error) {
+			captured = request.(*metadataapi.Request)
+			return &metadataapi.Response{
+				Brokers: []metadataapi.ResponseBroker{{NodeID: 1, Host: "kafka-1", Port: 9092}},
+				Topics: []metadataapi.ResponseTopic{{
+					Name: "orders",
+					Partitions: []metadataapi.ResponsePartition{{
+						PartitionIndex: 0,
+						LeaderID:       1,
+					}},
+				}},
+			}, nil
+		}),
+	}
+	if err := probeKafkaMetadata(context.Background(), client, "orders"); err != nil {
+		t.Fatal(err)
+	}
+	if captured == nil || len(captured.TopicNames) != 1 || captured.TopicNames[0] != "orders" {
+		t.Fatalf("metadata request = %#v", captured)
+	}
+	if captured.AllowAutoTopicCreation {
+		t.Fatal("subscription metadata lookup enabled automatic topic creation")
+	}
+}
+
+func TestProbeKafkaMetadataRequestsBrokersWithoutListingTopics(t *testing.T) {
+	var captured *metadataapi.Request
+	client := &kafkago.Client{
+		Addr: kafkago.TCP("kafka-1:9092"),
+		Transport: kafkaRoundTripperFunc(func(_ context.Context, _ net.Addr, request kafkago.Request) (kafkago.Response, error) {
+			captured = request.(*metadataapi.Request)
+			return &metadataapi.Response{Brokers: []metadataapi.ResponseBroker{{NodeID: 1, Host: "kafka-1", Port: 9092}}}, nil
+		}),
+	}
+	if err := probeKafkaMetadata(context.Background(), client, ""); err != nil {
+		t.Fatal(err)
+	}
+	if captured == nil || captured.TopicNames == nil || len(captured.TopicNames) != 0 {
+		t.Fatalf("broker metadata request topics = %#v", captured)
+	}
+}
+
+func TestKafkaTopicMetadataErrorDoesNotMarkClusterUnavailable(t *testing.T) {
+	metadata := &kafkago.Client{
+		Addr: kafkago.TCP("kafka-1:9092"),
+		Transport: kafkaRoundTripperFunc(func(_ context.Context, _ net.Addr, _ kafkago.Request) (kafkago.Response, error) {
+			return &metadataapi.Response{
+				Brokers: []metadataapi.ResponseBroker{{NodeID: 1, Host: "kafka-1", Port: 9092}},
+				Topics:  []metadataapi.ResponseTopic{{Name: "missing", ErrorCode: int16(kafkago.UnknownTopicOrPartition)}},
+			}, nil
+		}),
+	}
+	client := &Kafka{
+		config:        defaultKafkaConfig(KafkaConfig{}),
+		dialer:        &kafkago.Dialer{},
+		metadata:      metadata,
+		subscriptions: make(map[string]*kafkaSubscription),
+	}
+	client.markUnavailable()
+	err := client.Subscribe(context.Background(), "missing", AtLeastOnce, func(context.Context, Message) {})
+	if err == nil {
+		t.Fatal("missing kafka topic was accepted")
+	}
+	if !client.Connected() || client.Reconnecting() {
+		t.Fatalf("topic error changed cluster state: connected=%t reconnecting=%t", client.Connected(), client.Reconnecting())
+	}
+}
@@ -27,6 +27,9 @@ func NormalizeSubscriptionSet(set SubscriptionSet) (SubscriptionSet, error) {
 		if set.Topics[index].QoS > ExactlyOnce {
 			return SubscriptionSet{}, fmt.Errorf("invalid mq qos %d for topic %q", set.Topics[index].QoS, set.Topics[index].Topic)
 		}
+		if set.Provider != ProviderEMQX && set.Topics[index].QoS > AtLeastOnce {
+			return SubscriptionSet{}, fmt.Errorf("mq provider %q supports qos 0 or 1 for topic %q", set.Provider, set.Topics[index].Topic)
+		}
 		if set.Topics[index].Handler == nil {
 			return SubscriptionSet{}, fmt.Errorf("mq subscription handler is nil for topic %q", set.Topics[index].Topic)
 		}
@@ -16,7 +16,7 @@ func TestNormalizeSubscriptionSetValidDeclaration(t *testing.T) {
 		Topics: []TopicSubscription{
 			{Topic: "  orders.created  ", QoS: AtMostOnce, Handler: handler},
 			{Topic: "orders.updated", QoS: AtLeastOnce, Handler: handler},
-			{Topic: "orders.deleted", QoS: ExactlyOnce, Handler: handler},
+			{Topic: "orders.deleted", QoS: AtLeastOnce, Handler: handler},
 		},
 	}

@@ -31,7 +31,7 @@ func TestNormalizeSubscriptionSetValidDeclaration(t *testing.T) {
 		Topics: []TopicSubscription{
 			{Topic: "orders.created", QoS: AtMostOnce, Handler: handler},
 			{Topic: "orders.updated", QoS: AtLeastOnce, Handler: handler},
-			{Topic: "orders.deleted", QoS: ExactlyOnce, Handler: handler},
+			{Topic: "orders.deleted", QoS: AtLeastOnce, Handler: handler},
 		},
 	}
 	if got.Owner != want.Owner || got.Provider != want.Provider {
@@ -53,6 +53,34 @@ func TestNormalizeSubscriptionSetValidDeclaration(t *testing.T) {
 	}
 }

+func TestNormalizeSubscriptionSetAcceptsMQTTQoS2(t *testing.T) {
+	set, err := NormalizeSubscriptionSet(SubscriptionSet{
+		Owner:    "devices",
+		Provider: ProviderEMQX,
+		Topics:   []TopicSubscription{{Topic: "devices/status", QoS: ExactlyOnce, Handler: func(context.Context, Message) {}}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if set.Topics[0].QoS != ExactlyOnce {
+		t.Fatalf("qos = %d, want %d", set.Topics[0].QoS, ExactlyOnce)
+	}
+}
+
+func TestNormalizeSubscriptionSetAcceptsKafka(t *testing.T) {
+	set, err := NormalizeSubscriptionSet(SubscriptionSet{
+		Owner:    "orders",
+		Provider: " KAFKA ",
+		Topics:   []TopicSubscription{{Topic: "orders.created", QoS: AtLeastOnce, Handler: func(context.Context, Message) {}}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if set.Provider != ProviderKafka {
+		t.Fatalf("provider = %q, want %q", set.Provider, ProviderKafka)
+	}
+}
+
 func TestNormalizeSubscriptionSetRejectsInvalidDeclarations(t *testing.T) {
 	handler := func(context.Context, Message) {}
 	base := func() SubscriptionSet {
@@ -82,7 +110,7 @@ func TestNormalizeSubscriptionSetRejectsInvalidDeclarations(t *testing.T) {
 		},
 		{
 			name:    "unsupported provider",
-			set:     func() SubscriptionSet { set := base(); set.Provider = "kafka"; return set }(),
+			set:     func() SubscriptionSet { set := base(); set.Provider = "pulsar"; return set }(),
 			wantErr: "unsupported mq provider",
 		},
 		{
@@ -128,6 +156,26 @@ func TestNormalizeSubscriptionSetRejectsInvalidDeclarations(t *testing.T) {
 			}(),
 			wantErr: "invalid mq qos",
 		},
+		{
+			name: "kafka qos 2",
+			set: func() SubscriptionSet {
+				set := base()
+				set.Provider = ProviderKafka
+				set.Topics[0].QoS = ExactlyOnce
+				return set
+			}(),
+			wantErr: "supports qos 0 or 1",
+		},
+		{
+			name: "rabbitmq qos 2",
+			set: func() SubscriptionSet {
+				set := base()
+				set.Provider = ProviderRabbitMQ
+				set.Topics[0].QoS = ExactlyOnce
+				return set
+			}(),
+			wantErr: "supports qos 0 or 1",
+		},
 	}

 	for _, test := range tests {