From 0b8e5b6bf8767b837dbae1791513efc5cbe7c3c7 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:49:31 +0200 Subject: [PATCH 1/7] Add NATS JetStream provider --- nats_jetstream_provider.go | 232 +++++++++++++++++++++++++++++++++++++ 1 file changed, 232 insertions(+) create mode 100644 nats_jetstream_provider.go diff --git a/nats_jetstream_provider.go b/nats_jetstream_provider.go new file mode 100644 index 0000000..7204c43 --- /dev/null +++ b/nats_jetstream_provider.go @@ -0,0 +1,232 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/nats-io/nats.go" +) + +// NATSJetStreamProviderConfig configures a NATS JetStream provider. +// +// Conn is injected and remains owned by the caller. The provider creates +// JetStream subscriptions lazily in Consume and never closes Conn. +type NATSJetStreamProviderConfig struct { + Conn *nats.Conn + + // Subject is the subscription subject. It defaults to SubjectPrefix + ">". + Subject string + // SubjectPrefix is prepended to the event projection on publish. + // It defaults to "events.". + SubjectPrefix string + // Durable configures a durable JetStream consumer. + Durable string + // Queue enables queue-group consumption for multiple consumers. + Queue string + + Codec EventCodec +} + +// NATSJetStreamProvider publishes to and consumes from NATS JetStream. +// Messages are acknowledged only after EventConsumer succeeds. +type NATSJetStreamProvider struct { + conn *nats.Conn + js nats.JetStreamContext + subject string + subjectPrefix string + durable string + queue string + codec EventCodec + + mu sync.Mutex + sub *nats.Subscription + closed atomic.Bool +} + +// WithNATSJetStream configures NATS JetStream as the EventStore provider. +func WithNATSJetStream(config NATSJetStreamProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { + return NewNATSJetStreamProvider(config) + }) +} + +// NewNATSJetStreamProvider creates a NATS JetStream provider without taking +// ownership of the injected connection. +func NewNATSJetStreamProvider(config NATSJetStreamProviderConfig) (*NATSJetStreamProvider, error) { + if config.Conn == nil { + return nil, errors.New("goeventbus: NATS connection must not be nil") + } + if config.SubjectPrefix == "" { + config.SubjectPrefix = "events." + } + if config.Subject == "" { + config.Subject = config.SubjectPrefix + ">" + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + + js, err := config.Conn.JetStream() + if err != nil { + return nil, fmt.Errorf("goeventbus: create NATS JetStream context: %w", err) + } + + return &NATSJetStreamProvider{ + conn: config.Conn, + js: js, + subject: config.Subject, + subjectPrefix: config.SubjectPrefix, + durable: config.Durable, + queue: config.Queue, + codec: config.Codec, + }, nil +} + +// Publish publishes an event to SubjectPrefix + Projection. +func (p *NATSJetStreamProvider) Publish(ctx context.Context, e Event) error { + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + projection, err := remoteProjection(e) + if err != nil { + return err + } + payload, err := p.codec.Encode(e) + if err != nil { + return err + } + + msg := &nats.Msg{ + Subject: p.subjectPrefix + projection, + Data: payload, + Header: nats.Header{}, + } + if e.ID != "" { + msg.Header.Set(nats.MsgIdHdr, e.ID) + } + + if _, err := p.js.PublishMsg(msg, nats.Context(ctx)); err != nil { + return fmt.Errorf("goeventbus: publish NATS JetStream event: %w", err) + } + return nil +} + +// Consume subscribes with explicit acknowledgements. Handler failures NAK the +// message so JetStream can redeliver it. +func (p *NATSJetStreamProvider) Consume(ctx context.Context, consumer EventConsumer) error { + if consumer == nil { + return ErrNilConsumer + } + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + + sub, err := p.subscribe() + if err != nil { + return err + } + defer func() { + p.mu.Lock() + if p.sub == sub { + p.sub = nil + } + p.mu.Unlock() + _ = sub.Unsubscribe() + }() + + for { + if p.closed.Load() { + return ErrProviderClosed + } + msg, err := sub.NextMsgWithContext(ctx) + if err != nil { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return ctx.Err() + } + if p.closed.Load() { + return ErrProviderClosed + } + return fmt.Errorf("goeventbus: consume NATS JetStream event: %w", err) + } + + event, err := p.codec.Decode(msg.Data) + if err != nil { + _ = msg.Nak() + return fmt.Errorf("goeventbus: decode NATS JetStream event: %w", err) + } + if err := consumer(ctx, event); err != nil { + _ = msg.Nak() + return err + } + if err := msg.Ack(); err != nil { + return fmt.Errorf("goeventbus: acknowledge NATS JetStream event: %w", err) + } + } +} + +func (p *NATSJetStreamProvider) subscribe() (*nats.Subscription, error) { + p.mu.Lock() + defer p.mu.Unlock() + + if p.closed.Load() { + return nil, ErrProviderClosed + } + if p.sub != nil && p.sub.IsValid() { + return nil, errors.New("goeventbus: NATS JetStream provider already has an active consumer") + } + + opts := []nats.SubOpt{nats.ManualAck(), nats.AckExplicit()} + if p.durable != "" { + opts = append(opts, nats.Durable(p.durable)) + } + + var ( + sub *nats.Subscription + err error + ) + if strings.TrimSpace(p.queue) != "" { + sub, err = p.js.QueueSubscribeSync(p.subject, p.queue, opts...) + } else { + sub, err = p.js.SubscribeSync(p.subject, opts...) + } + if err != nil { + return nil, fmt.Errorf("goeventbus: subscribe NATS JetStream subject %q: %w", p.subject, err) + } + p.sub = sub + return sub, nil +} + +// Close stops active consumption. The injected NATS connection is not closed. +func (p *NATSJetStreamProvider) Close() error { + if p.closed.Swap(true) { + return nil + } + p.mu.Lock() + defer p.mu.Unlock() + if p.sub == nil { + return nil + } + err := p.sub.Unsubscribe() + p.sub = nil + if err != nil && !errors.Is(err, nats.ErrBadSubscription) { + return fmt.Errorf("goeventbus: close NATS JetStream subscription: %w", err) + } + return nil +} + +// keep time imported for compatibility with nats context implementations that +// surface timeout errors through NextMsgWithContext on older clients. +var _ = time.Second + +var _ Provider = (*NATSJetStreamProvider)(nil) From 4af3dab36142ad7c244eafe35f55c57966cfbb43 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:49:33 +0200 Subject: [PATCH 2/7] Add Kafka provider --- kafka_provider.go | 197 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 197 insertions(+) create mode 100644 kafka_provider.go diff --git a/kafka_provider.go b/kafka_provider.go new file mode 100644 index 0000000..d29b54d --- /dev/null +++ b/kafka_provider.go @@ -0,0 +1,197 @@ +package GoEventBus + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + "time" + + "github.com/segmentio/kafka-go" +) + +// KafkaProviderConfig configures an Apache Kafka provider. +type KafkaProviderConfig struct { + Brokers []string + Topic string + Group string + + // Partition pins publishing/consumption to a partition when >= 0. + // Partition must remain -1 when Group is configured because Kafka assigns + // partitions to consumer-group members. + Partition int + + MinBytes int + MaxBytes int + MaxWait time.Duration + Codec EventCodec +} + +// KafkaProvider publishes through kafka-go Writer and consumes with synchronous +// offset commits after successful event handling. +type KafkaProvider struct { + writer *kafka.Writer + reader *kafka.Reader + + topic string + partition int + codec EventCodec + + closeOnce sync.Once + closeErr error + closed atomic.Bool +} + +// WithKafka configures Apache Kafka as the EventStore provider. +func WithKafka(config KafkaProviderConfig) EventStoreOption { + return withProviderFactory(func() (Provider, error) { + return NewKafkaProvider(config) + }) +} + +// NewKafkaProvider builds Kafka clients. Network I/O begins on Publish/Consume. +func NewKafkaProvider(config KafkaProviderConfig) (*KafkaProvider, error) { + if len(config.Brokers) == 0 { + return nil, errors.New("goeventbus: Kafka brokers must not be empty") + } + if config.Topic == "" { + return nil, errors.New("goeventbus: Kafka topic must not be empty") + } + if config.Partition < -1 { + return nil, errors.New("goeventbus: Kafka partition must be -1 or greater") + } + if config.Group != "" && config.Partition >= 0 { + return nil, errors.New("goeventbus: Kafka partition cannot be fixed when a consumer group is configured") + } + if config.MinBytes <= 0 { + config.MinBytes = 1 + } + if config.MaxBytes <= 0 { + config.MaxBytes = 10e6 + } + if config.MaxWait <= 0 { + config.MaxWait = time.Second + } + if config.Codec == nil { + config.Codec = JSONCodec{} + } + + writer := &kafka.Writer{ + Addr: kafka.TCP(config.Brokers...), + Topic: config.Topic, + Balancer: &kafka.Hash{}, + RequiredAcks: kafka.RequireAll, + Async: false, + } + + readerConfig := kafka.ReaderConfig{ + Brokers: config.Brokers, + GroupID: config.Group, + Topic: config.Topic, + MinBytes: config.MinBytes, + MaxBytes: config.MaxBytes, + MaxWait: config.MaxWait, + CommitInterval: 0, + } + if config.Group == "" && config.Partition >= 0 { + readerConfig.Partition = config.Partition + } + + return &KafkaProvider{ + writer: writer, + reader: kafka.NewReader(readerConfig), + topic: config.Topic, + partition: config.Partition, + codec: config.Codec, + }, nil +} + +// Publish writes one Kafka message. Projection is stored as the message key to +// preserve per-projection ordering when the default hash balancer is used. +func (p *KafkaProvider) Publish(ctx context.Context, e Event) error { + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + projection, err := remoteProjection(e) + if err != nil { + return err + } + payload, err := p.codec.Encode(e) + if err != nil { + return err + } + + msg := kafka.Message{ + Topic: p.topic, + Key: []byte(projection), + Value: payload, + Time: time.Now().UTC(), + } + if p.partition >= 0 { + msg.Partition = p.partition + } + if e.ID != "" { + msg.Headers = append(msg.Headers, kafka.Header{Key: "goeventbus-event-id", Value: []byte(e.ID)}) + } + + if err := p.writer.WriteMessages(ctx, msg); err != nil { + return fmt.Errorf("goeventbus: publish Kafka event: %w", err) + } + return nil +} + +// Consume fetches one message at a time and commits its offset only after the +// EventConsumer returns nil. Failed handling leaves the offset uncommitted. +func (p *KafkaProvider) Consume(ctx context.Context, consumer EventConsumer) error { + if consumer == nil { + return ErrNilConsumer + } + if p.closed.Load() { + return ErrProviderClosed + } + if ctx == nil { + ctx = context.Background() + } + + for { + if p.closed.Load() { + return ErrProviderClosed + } + msg, err := p.reader.FetchMessage(ctx) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + if p.closed.Load() { + return ErrProviderClosed + } + return fmt.Errorf("goeventbus: fetch Kafka event: %w", err) + } + + event, err := p.codec.Decode(msg.Value) + if err != nil { + return fmt.Errorf("goeventbus: decode Kafka event at partition %d offset %d: %w", msg.Partition, msg.Offset, err) + } + if err := consumer(ctx, event); err != nil { + return err + } + if err := p.reader.CommitMessages(ctx, msg); err != nil { + return fmt.Errorf("goeventbus: commit Kafka offset %d: %w", msg.Offset, err) + } + } +} + +// Close closes the owned Kafka reader and writer. +func (p *KafkaProvider) Close() error { + p.closeOnce.Do(func() { + p.closed.Store(true) + p.closeErr = errors.Join(p.reader.Close(), p.writer.Close()) + }) + return p.closeErr +} + +var _ Provider = (*KafkaProvider)(nil) From a8d4e22f195f2d8b295a87f7005bf0bcffce3e36 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:49:37 +0200 Subject: [PATCH 3/7] Test new broker options --- broker_options_test.go | 53 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) create mode 100644 broker_options_test.go diff --git a/broker_options_test.go b/broker_options_test.go new file mode 100644 index 0000000..b1ee187 --- /dev/null +++ b/broker_options_test.go @@ -0,0 +1,53 @@ +package GoEventBus + +import ( + "errors" + "testing" +) + +func TestWithNATSJetStreamRejectsNilConnectionLazily(t *testing.T) { + var dispatcher Dispatcher + store := NewEventStore(&dispatcher, 8, Block, WithNATSJetStream(NATSJetStreamProviderConfig{})) + err := store.PublishToProvider(nil, Event{Projection: "order.created"}) + if err == nil || err.Error() != "goeventbus: NATS connection must not be nil" { + t.Fatalf("expected nil NATS connection error, got %v", err) + } +} + +func TestWithKafkaValidatesConfigurationLazily(t *testing.T) { + var dispatcher Dispatcher + store := NewEventStore(&dispatcher, 8, Block, WithKafka(KafkaProviderConfig{})) + err := store.PublishToProvider(nil, Event{Projection: "order.created"}) + if err == nil || err.Error() != "goeventbus: Kafka brokers must not be empty" { + t.Fatalf("expected Kafka brokers error, got %v", err) + } +} + +func TestKafkaRejectsFixedPartitionWithGroup(t *testing.T) { + _, err := NewKafkaProvider(KafkaProviderConfig{ + Brokers: []string{"localhost:9092"}, + Topic: "events", + Group: "billing", + Partition: 0, + }) + if err == nil { + t.Fatal("expected group/partition validation error") + } +} + +func TestKafkaRejectsNonStringProjectionBeforeNetwork(t *testing.T) { + provider, err := NewKafkaProvider(KafkaProviderConfig{ + Brokers: []string{"localhost:9092"}, + Topic: "events", + Partition: -1, + }) + if err != nil { + t.Fatal(err) + } + defer provider.Close() + + err = provider.Publish(nil, Event{Projection: 123}) + if !errors.Is(err, ErrStringProjection) { + t.Fatalf("expected ErrStringProjection, got %v", err) + } +} From 9ae5bcc96b957c35d9ebf389a94ae6965d1f8a97 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:49:41 +0200 Subject: [PATCH 4/7] Add NATS and Kafka dependencies --- go.mod | 2 ++ 1 file changed, 2 insertions(+) diff --git a/go.mod b/go.mod index fb1265d..635f0d1 100644 --- a/go.mod +++ b/go.mod @@ -6,8 +6,10 @@ toolchain go1.23.7 require ( github.com/fasthttp/router v1.5.4 + github.com/nats-io/nats.go v1.39.1 github.com/rabbitmq/amqp091-go v1.14.0 github.com/redis/go-redis/v9 v9.18.0 + github.com/segmentio/kafka-go v0.4.48 github.com/testcontainers/testcontainers-go v0.34.0 github.com/valyala/fasthttp v1.61.0 ) From c4c70927c039d6e4acb9aa1997800af11f9eefbc Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:50:59 +0200 Subject: [PATCH 5/7] Fix Kafka partition configuration --- kafka_provider.go | 53 ++++++++++++++++++++++++++++++++--------------- 1 file changed, 36 insertions(+), 17 deletions(-) diff --git a/kafka_provider.go b/kafka_provider.go index d29b54d..222ce87 100644 --- a/kafka_provider.go +++ b/kafka_provider.go @@ -17,10 +17,10 @@ type KafkaProviderConfig struct { Topic string Group string - // Partition pins publishing/consumption to a partition when >= 0. - // Partition must remain -1 when Group is configured because Kafka assigns - // partitions to consumer-group members. - Partition int + // Partition pins publishing and non-group consumption to one partition. + // Leave nil to let Kafka select/assign partitions. Partition and Group are + // mutually exclusive for consumers because group membership owns assignment. + Partition *int MinBytes int MaxBytes int @@ -35,7 +35,7 @@ type KafkaProvider struct { reader *kafka.Reader topic string - partition int + partition *int codec EventCodec closeOnce sync.Once @@ -58,10 +58,10 @@ func NewKafkaProvider(config KafkaProviderConfig) (*KafkaProvider, error) { if config.Topic == "" { return nil, errors.New("goeventbus: Kafka topic must not be empty") } - if config.Partition < -1 { - return nil, errors.New("goeventbus: Kafka partition must be -1 or greater") + if config.Partition != nil && *config.Partition < 0 { + return nil, errors.New("goeventbus: Kafka partition must be zero or greater") } - if config.Group != "" && config.Partition >= 0 { + if config.Group != "" && config.Partition != nil { return nil, errors.New("goeventbus: Kafka partition cannot be fixed when a consumer group is configured") } if config.MinBytes <= 0 { @@ -77,10 +77,15 @@ func NewKafkaProvider(config KafkaProviderConfig) (*KafkaProvider, error) { config.Codec = JSONCodec{} } + var balancer kafka.Balancer = &kafka.Hash{} + if config.Partition != nil { + balancer = fixedKafkaPartitionBalancer(*config.Partition) + } + writer := &kafka.Writer{ Addr: kafka.TCP(config.Brokers...), Topic: config.Topic, - Balancer: &kafka.Hash{}, + Balancer: balancer, RequiredAcks: kafka.RequireAll, Async: false, } @@ -94,8 +99,8 @@ func NewKafkaProvider(config KafkaProviderConfig) (*KafkaProvider, error) { MaxWait: config.MaxWait, CommitInterval: 0, } - if config.Group == "" && config.Partition >= 0 { - readerConfig.Partition = config.Partition + if config.Partition != nil { + readerConfig.Partition = *config.Partition } return &KafkaProvider{ @@ -107,8 +112,9 @@ func NewKafkaProvider(config KafkaProviderConfig) (*KafkaProvider, error) { }, nil } -// Publish writes one Kafka message. Projection is stored as the message key to -// preserve per-projection ordering when the default hash balancer is used. +// Publish writes one Kafka message. Projection is stored as the message key, +// preserving per-projection ordering with the default hash balancer. A fixed +// partition, when configured, overrides the balancer. func (p *KafkaProvider) Publish(ctx context.Context, e Event) error { if p.closed.Load() { return ErrProviderClosed @@ -131,9 +137,6 @@ func (p *KafkaProvider) Publish(ctx context.Context, e Event) error { Value: payload, Time: time.Now().UTC(), } - if p.partition >= 0 { - msg.Partition = p.partition - } if e.ID != "" { msg.Headers = append(msg.Headers, kafka.Header{Key: "goeventbus-event-id", Value: []byte(e.ID)}) } @@ -145,7 +148,8 @@ func (p *KafkaProvider) Publish(ctx context.Context, e Event) error { } // Consume fetches one message at a time and commits its offset only after the -// EventConsumer returns nil. Failed handling leaves the offset uncommitted. +// EventConsumer returns nil. Failed handling leaves the offset uncommitted for +// redelivery according to Kafka consumer-group semantics. func (p *KafkaProvider) Consume(ctx context.Context, consumer EventConsumer) error { if consumer == nil { return ErrNilConsumer @@ -194,4 +198,19 @@ func (p *KafkaProvider) Close() error { return p.closeErr } +type fixedKafkaPartitionBalancer int + +func (b fixedKafkaPartitionBalancer) Balance(_ kafka.Message, partitions ...int) int { + want := int(b) + for _, partition := range partitions { + if partition == want { + return want + } + } + if len(partitions) == 0 { + return 0 + } + return partitions[0] +} + var _ Provider = (*KafkaProvider)(nil) From a6fee4bb80485f4677742e0722432d0a79ab74b1 Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:51:02 +0200 Subject: [PATCH 6/7] Cover Kafka group and partition configuration --- broker_options_test.go | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/broker_options_test.go b/broker_options_test.go index b1ee187..5f1ffd5 100644 --- a/broker_options_test.go +++ b/broker_options_test.go @@ -24,22 +24,34 @@ func TestWithKafkaValidatesConfigurationLazily(t *testing.T) { } func TestKafkaRejectsFixedPartitionWithGroup(t *testing.T) { + partition := 0 _, err := NewKafkaProvider(KafkaProviderConfig{ Brokers: []string{"localhost:9092"}, Topic: "events", Group: "billing", - Partition: 0, + Partition: &partition, }) if err == nil { t.Fatal("expected group/partition validation error") } } +func TestKafkaGroupUsesZeroValueConfigWithoutPartitionConflict(t *testing.T) { + provider, err := NewKafkaProvider(KafkaProviderConfig{ + Brokers: []string{"localhost:9092"}, + Topic: "events", + Group: "billing", + }) + if err != nil { + t.Fatalf("expected group config to be valid, got %v", err) + } + _ = provider.Close() +} + func TestKafkaRejectsNonStringProjectionBeforeNetwork(t *testing.T) { provider, err := NewKafkaProvider(KafkaProviderConfig{ - Brokers: []string{"localhost:9092"}, - Topic: "events", - Partition: -1, + Brokers: []string{"localhost:9092"}, + Topic: "events", }) if err != nil { t.Fatal(err) From bcba36ea12f0bc7be4f03089bc5f1db4ad4af3fb Mon Sep 17 00:00:00 2001 From: Kamil Mosciszko Date: Mon, 5 Oct 2026 19:51:22 +0200 Subject: [PATCH 7/7] Document NATS JetStream and Kafka providers --- README.MD | 46 +++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 45 insertions(+), 1 deletion(-) diff --git a/README.MD b/README.MD index 268a987..b4a1841 100644 --- a/README.MD +++ b/README.MD @@ -63,7 +63,7 @@ That gives you intelligent routing without putting an LLM call in the `Publish() - **Lifecycle hooks** with `OnBefore`, `OnAfter`, and `OnError`. - **Back-pressure policies**: `DropOldest`, `Block`, or `ReturnError`. - **Dead-letter queue** with replay support and panic recovery. -- **Redis Streams and RabbitMQ EventStore options** for cross-process delivery. +- **Redis Streams, RabbitMQ, NATS JetStream, and Kafka EventStore options** for cross-process delivery. - **Transactions** with local buffering, transport-aware commit, and optional Rules/Cache/Jev event selection. - **Scheduling** with absolute/relative timers and optional Rules/Cache/Jev event selection. - **Metrics** for published, processed, and failed events. @@ -718,6 +718,50 @@ if err != nil { fmt.Println("published decision:", decision.Choice) ``` +## NATS JetStream + +```go +nc, err := nats.Connect(nats.DefaultURL) +if err != nil { + log.Fatal(err) +} + +store := GoEventBus.NewEventStore( + &dispatcher, + 1024, + GoEventBus.Block, + GoEventBus.WithNATSJetStream(GoEventBus.NATSJetStreamProviderConfig{ + Conn: nc, + SubjectPrefix: "events.", + Subject: "events.>", + Durable: "billing", + Queue: "billing-workers", + }), +) +``` + +The injected NATS connection remains application-owned. JetStream messages use +explicit acknowledgements; handler failures are NAKed for broker redelivery. + +## Apache Kafka + +```go +store := GoEventBus.NewEventStore( + &dispatcher, + 1024, + GoEventBus.Block, + GoEventBus.WithKafka(GoEventBus.KafkaProviderConfig{ + Brokers: []string{"localhost:9092"}, + Topic: "events", + Group: "billing", + }), +) +``` + +Kafka consumer offsets are committed synchronously only after successful event +handling. With the default hash balancer, the string projection is used as the +message key, preserving ordering for a projection within a partition. + Providers use `JSONCodec` by default. Supply a custom `DecodePayload` when consumers need concrete payload types rather than generic JSON values.