diff --git a/.agents/skills/fkafka/SKILL.md b/.agents/skills/fkafka/SKILL.md new file mode 100644 index 0000000..9e1f880 --- /dev/null +++ b/.agents/skills/fkafka/SKILL.md @@ -0,0 +1,66 @@ +--- +name: fkafka +description: Use whenever generating or reviewing F# code that produces to or consumes from Apache Kafka via the Alma.Kafka library — calls to Consumer.consume, ConsumerConfiguration.createWithConnection, Producer.create / Producer.produce, MessageToProduce, Admin.lags, Checker, or composes TracedMessage handling, manual/auto commit (CommitMessage), external offset checkpoints (GetCheckpoint), or B3 trace propagation through Kafka headers. Trigger also on mentions of BrokerList, StreamName, GroupId, consumer lag, KSQL message keys, MessageKey.Delimited, or "read/write a Kafka topic in F#". +--- + +# F-Kafka + +Library: [alma-oss/fkafka](https://github.com/alma-oss/fkafka) +NuGet: `Alma.Kafka` + +## Purpose + +`Alma.Kafka` is an F# wrapper over `Confluent.Kafka` for producing and consuming messages to/from Kafka topics. It exposes a typed, railway-oriented (`AsyncResult`) API with consumer-lag monitoring, cluster/topic health checking, manual or automatic offset commit, external offset checkpoints, and automatic distributed-trace propagation through Kafka message headers. + +## When to Use + +- Consuming a Kafka topic as a lazy F# sequence of typed messages. +- Producing messages with a key (simple or KSQL-style delimited) and custom headers. +- Monitoring consumer lag per partition for a given group. +- Adding health checks (cluster/topic availability with retry) around producing/consuming. +- Storing offsets in external storage instead of Kafka's built-in commit. + +## When NOT to Use + +- Non-Kafka message brokers. +- Low-level partition assignment or admin operations beyond topic listing and lag — drop to `Confluent.Kafka` directly. +- Schema-registry / Avro / Protobuf serialization — this library works with `string` payloads. + +## Main Concepts + +- `BrokerList` — single-case DU wrapping the comma-separated bootstrap-server string. +- `StreamName` — topic name; either an explicit `StreamName` or derived from an `Instance`. +- `GroupId` — `Id of string` for a stable group, or `Random` for a unique throwaway group (always reads from the beginning). +- `ConnectionConfiguration` — record of `{ BrokerList; Topic }`. +- `ConsumerConfiguration` — full consumer setup; build with `ConsumerConfiguration.createWithConnection` or `createWithDefaults`. +- `Consumer.consume` — turns a configuration into a lazy, effectively infinite `seq` of `TracedMessage`. +- `TracedMessage<'Message>` — `{ Commit; Message; Trace }`; map the payload via `TracedMessage.map`, finish the span via `TracedMessage.finish`. +- `CommitMessage` — `Automatically` (Kafka autocommit) or `Manually of FailOnNotCommittedMessage`. +- `ManualCommit` — handle whose `ManualCommit.execute` commits the current offset under manual mode. +- `GetCheckpoint` — optional function to resolve a starting offset per partition from external storage. +- `ProducerConfiguration` / `Producer` — producer setup and the disposable producer handle. +- `MessageToProduce` — `{ Key; Headers; Value }`; build with `MessageToProduce.create` / `createWithHeaders`. +- `MessageKey` — `Simple of string` or `Delimited of string list` (joined with `,`, spaces stripped, KSQL-compatible). +- `Admin` — cluster inspection: `createAdmin`, `getAllTopics`, `topicExists`, `isUp`, `lags`. +- `PartitionLag` — `{ Partition; Lag }` produced by `Admin.lags`. +- `Checker` / `IntervalChecker` — health-check records with `defaultChecker` presets. +- `Event<'KeyData,'MetaData,'DomainData>` / `CommonEvent` — typed event-envelope schema with `EventId`, `CorrelationId`, `CausationId`, `Resource`. +- `MetaData` — parsed message metadata (`OnlyCreatedAt` or `CreatedAndProcessed`) via `MetaData.parse`. + +## Related Libraries + +- `Confluent.Kafka` — underlying client; its types surface in errors and handles. +- `Feather.ErrorHandling` — `AsyncResult` / `asyncResult` CE used across the API. +- `Alma.Tracing` — span types and B3 header inject/extract used for trace propagation. +- `Alma.ServiceIdentification` — `Instance`, `Domain`, `Context` used by `StreamName` and event envelopes. +- `Alma.Metrics` — service status (`MarkAsEnabled` / `MarkAsDisabled`) wired into health checks. + +## Keywords for Search + +Kafka, Alma.Kafka, fkafka, F# Kafka, Confluent.Kafka, consumer, producer, BrokerList, StreamName, GroupId, ConsumerConfiguration, Consumer.consume, TracedMessage, CommitMessage, ManualCommit, GetCheckpoint, checkpoint, offset, consumer lag, Admin.lags, PartitionLag, Producer, MessageToProduce, MessageKey, Delimited, KSQL, Header, Checker, IntervalChecker, health check, trace propagation, B3 headers, Event, MetaData + +## Reference Files + +- For composition principles and recommended API usage, read `references/preferred-patterns.md`. +- For known pitfalls and incorrect assumptions, read `references/anti-patterns.md`. +- For worked code examples, read `references/examples.md`. diff --git a/.agents/skills/fkafka/references/anti-patterns.md b/.agents/skills/fkafka/references/anti-patterns.md new file mode 100644 index 0000000..01e4f6a --- /dev/null +++ b/.agents/skills/fkafka/references/anti-patterns.md @@ -0,0 +1,55 @@ +# Anti-Patterns — Alma.Kafka + +Each entry is **mistake → why → fix**. + +## Lifecycle + +- **Mistake:** Creating a `Producer` or `Consumer` without `use` (or never calling `Producer.close`). + **Why:** Both hold native `librdkafka` resources and implement `IDisposable`; a leaked producer may never flush buffered messages, and a leaked consumer never closes its group session. + **Fix:** Bind with `use producer = Producer.create config`, or scope the `Consumer.consume` sequence so disposal happens when the scope ends. + +- **Mistake:** Forcing the consume sequence with `Seq.toList` / `List.ofSeq`. + **Why:** `Consumer.consume` is effectively infinite; materializing it blocks forever. + **Fix:** Stream it lazily with `Seq.iter` / `Seq.map` / `Seq.truncate`. + +## Group IDs + +- **Mistake:** Using `GroupId.Random` in production. + **Why:** It generates a unique group id per process start, so the consumer always re-reads the topic from the beginning and never shares progress across instances. + **Fix:** Use `GroupId.Id ""` for any long-running consumer; reserve `GroupId.Random` for tests and one-off reads. + +## Commit Handling + +- **Mistake:** Selecting `CommitMessage.Manually` but never calling `ManualCommit.execute`. + **Why:** Offsets are never committed, so the consumer reprocesses from the last committed position on restart; with `FailOnNotCommittedMessage.WithException` the next consume fails with `ConsumeError.PreviousMessageWasNotCommited`. + **Fix:** After successfully processing a message, call `ManualCommit.execute message.Commit` and handle the `ManualCommitError` result. + +- **Mistake:** Discarding the `Result` from `ManualCommit.execute`. + **Why:** A failed commit (`ManualCommitError.KafkaException` / `RuntimeException`) is silently lost, masking duplicate-processing risk. + **Fix:** Pattern-match the result and log/propagate the error. + +## External Checkpoints + +- **Mistake:** Throwing from a `GetCheckpoint` function when no stored offset exists. + **Why:** The library treats a successful result with `Offset = None` as "no checkpoint" and falls back to the earliest offset; an exception instead aborts partition assignment. + **Fix:** Return `Ok { TopicPartition = tp; Offset = None }` when nothing is stored. + +- **Mistake:** Saving the external checkpoint and committing the Kafka offset as independent, non-atomic steps. + **Why:** A crash between the two leaves stored and committed offsets out of sync, causing message loss or duplication. + **Fix:** Persist the external checkpoint and the Kafka commit together (transactionally or with idempotent processing) after each message or batch. + +## Message Keys + +- **Mistake:** Embedding spaces or your own separators in a `MessageKey.Delimited` list. + **Why:** Delimited keys are joined with `,` and have spaces stripped to stay KSQL-compatible; manual separators or spaces produce keys that don't match downstream consumers. + **Fix:** Pass the raw field values as a `string list` and let `MessageKey.Delimited` build the comma-joined key. + +## Wrong Abstractions + +- **Mistake:** Reaching into the `Trace` module or hand-injecting B3 headers. + **Why:** Trace inject/extract is internal and runs automatically inside `Producer.produce` and `Consumer.consume`; duplicating it produces conflicting spans. + **Fix:** Rely on the built-in propagation; only create application-level child spans through `Alma.Tracing`. + +- **Mistake:** Pattern-matching wrapper DUs (`BrokerList`, `Offset`, `GroupId`) inline across modules. + **Why:** It bypasses the companion-module API and breaks if the representation changes. + **Fix:** Use the module accessors (`BrokerList.value`, `GroupId.value`, …). diff --git a/.agents/skills/fkafka/references/examples.md b/.agents/skills/fkafka/references/examples.md new file mode 100644 index 0000000..1231d7d --- /dev/null +++ b/.agents/skills/fkafka/references/examples.md @@ -0,0 +1,171 @@ +# Examples — Alma.Kafka + +This file is the single source of truth for all example code in this skill. Each example is self-contained and ordered by increasing complexity. Payloads are `string`; `parseMessage` is a placeholder for your own deserialization. + +## Basic Consume + +```fsharp +open Alma.Kafka + +let connection = { + BrokerList = BrokerList "127.0.0.1:9092" + Topic = StreamName "demo-topic" +} + +let configuration = ConsumerConfiguration.createWithConnection connection (GroupId.Id "demo-group") + +let parseMessage (raw: string) = raw // replace with real deserialization + +Consumer.consume configuration (TracedMessage.message >> parseMessage) +|> Seq.iter (fun message -> printfn "Message: %s" message) +``` + +## Produce With Key And Headers + +```fsharp +open Alma.Kafka + +let connection = { + BrokerList = BrokerList "127.0.0.1:9092" + Topic = StreamName "demo-topic" +} + +use producer = Producer.create (ProducerConfiguration.createWithConnection connection) + +let headers = [ + Header.ofString (HeaderKey "source") "web-api" +] + +// Simple key +MessageToProduce.createWithHeaders headers (MessageKey.Simple "entity-42", "payload") +|> Producer.produce producer + +// Composite, KSQL-compatible key (joined with ",") +MessageToProduce.create (MessageKey.Delimited [ "tenant-1"; "entity-42" ], "payload") +|> Producer.produceSingle producer // produce + flush +``` + +## Consume With Manual Commit + +```fsharp +open Alma.Kafka + +let connection = { + BrokerList = BrokerList "127.0.0.1:9092" + Topic = StreamName "demo-topic" +} + +let configuration = + { ConsumerConfiguration.createWithConnection connection (GroupId.Id "demo-group") with + CommitMessage = CommitMessage.Manually FailOnNotCommittedMessage.WithException + } + +Consumer.consume configuration id +|> Seq.iter (fun tracedMessage -> + // process tracedMessage.Message ... + match ManualCommit.execute tracedMessage.Commit with + | Ok () -> () + | Error (ManualCommitError.KafkaException e) -> eprintfn "commit failed: %A" e + | Error (ManualCommitError.RuntimeException e) -> eprintfn "commit failed: %A" e +) +``` + +## Consumer Lag + +```fsharp +open Microsoft.Extensions.Logging +open Alma.Kafka +open Alma.Kafka.Admin + +let runLag (logger: ILogger) = + let connection = { + BrokerList = BrokerList "127.0.0.1:9092" + Topic = StreamName "demo-topic" + } + + let totalLag = + Admin.lags logger connection (GroupId.Id "demo-group") + |> Async.RunSynchronously + |> List.sumBy PartitionLag.lag + + printfn "total lag: %d" totalLag +``` + +## External Checkpoint + +```fsharp +open Alma.Kafka +open Feather.ErrorHandling + +// Resolve a starting offset from external storage; return Offset = None when nothing is stored. +let getCheckpoint (groupId: GroupId) (topicPartition: TopicPartition): AsyncResult = asyncResult { + let! storedOffset = ExternalStore.tryGetOffset groupId topicPartition // your code: returns Offset option + return { TopicPartition = topicPartition; Offset = storedOffset } +} + +let connection = { + BrokerList = BrokerList "127.0.0.1:9092" + Topic = StreamName "demo-topic" +} + +let configuration = + { ConsumerConfiguration.createWithConnection connection (GroupId.Id "demo-group") with + CommitMessage = CommitMessage.Manually FailOnNotCommittedMessage.WithException + GetCheckpoint = Some getCheckpoint + } + +Consumer.consume configuration id +|> Seq.iter (fun tracedMessage -> + // process tracedMessage.Message, then persist the external checkpoint and commit atomically + match ManualCommit.execute tracedMessage.Commit with + | Ok () -> ExternalStore.saveOffset (GroupId.Id "demo-group") tracedMessage // your code + | Error e -> eprintfn "commit failed: %A" e +) +``` + +## Consume With Application Tracing + +```fsharp +open Alma.Kafka +open Alma.Tracing + +let connection = { + BrokerList = BrokerList "127.0.0.1:9092" + Topic = StreamName "demo-topic" +} + +let configuration = ConsumerConfiguration.createWithConnection connection (GroupId.Id "demo-group") + +// The consumer's own span is created automatically from the message's B3 headers. +// Here we start an application child span for processing and finish it ourselves. +Consumer.consume configuration (fun tracedMessage -> + tracedMessage.Message, + "Process message" |> Trace.ChildOf.start tracedMessage.Trace +) +|> Seq.iter (fun (message, processTrace) -> + // process message ... + processTrace |> Trace.finish +) +``` + +## Health Checked Producer + +```fsharp +open Alma.Kafka + +let connection = { + BrokerList = BrokerList "127.0.0.1:9092" + Topic = StreamName "demo-topic" +} + +// Blocks with retry until the cluster and topic are available, then yields a connected producer. +let configuration = + { ProducerConfiguration.createWithConnection connection with + Checker = Some Checker.defaultChecker + } + +use producer = Producer.create configuration + +MessageToProduce.create (MessageKey.Simple "entity-42", "payload") +|> Producer.produceSingle producer +``` diff --git a/.agents/skills/fkafka/references/preferred-patterns.md b/.agents/skills/fkafka/references/preferred-patterns.md new file mode 100644 index 0000000..19daca2 --- /dev/null +++ b/.agents/skills/fkafka/references/preferred-patterns.md @@ -0,0 +1,42 @@ +# Preferred Patterns — Alma.Kafka + +## Core Principles + +- Build configurations through the smart constructors (`ConsumerConfiguration.createWithConnection`, `ProducerConfiguration.createWithConnection`) and override individual fields with record-update syntax rather than constructing the records by hand. New optional fields get sane defaults this way. +- Treat every wrapper type through its companion module (`BrokerList.value`, `StreamName.value`, `GroupId.value`, `PartitionLag.lag`). Do not pattern-match the DU inline outside the module that owns it. +- Both `Producer` and `Consumer` own native resources and implement `IDisposable`. Bind them with `use` (or `Producer.close` / consume the sequence inside a scope) so the producer flushes and the consumer closes deterministically. +- `Consumer.consume` returns a lazy, effectively infinite sequence. Drive it with `Seq.iter` / `Seq.map` and let the surrounding scope control lifetime; do not force it with `Seq.toList`. + +## Recommended API Usage + +- Consuming: pass a projection from `TracedMessage` to your payload as the second argument of `Consumer.consume`. The simplest projection is `TracedMessage.message >> parseMessage`. See `examples.md` → Basic Consume. +- Producing: build the value with `MessageToProduce.create (key, value)` or `MessageToProduce.createWithHeaders headers (key, value)`, then `Producer.produce` (batch) or `Producer.produceSingle` (produce + flush). See `examples.md` → Produce With Key And Headers. +- Choose `MessageKey.Simple` for a single-field key and `MessageKey.Delimited` when the key is a composite that must stay KSQL-compatible. +- Lag monitoring: `Admin.lags` returns `PartitionLag list`; sum with `List.sumBy PartitionLag.lag`. See `examples.md` → Consumer Lag. + +## Error Handling + +- The API is railway-oriented: consume/commit results are `Result` / `AsyncResult` carrying DU error types (`ConsumeError`, `ManualCommitError`, `MetaDataParseError`). Match on the specific cases rather than catching exceptions. +- `ConsumeError.PreviousMessageWasNotCommited` only appears under manual commit with `FailOnNotCommittedMessage.WithException`; it signals the previous message was never committed. +- Under manual commit, always inspect the `Result` returned by `ManualCommit.execute` and surface `ManualCommitError` instead of ignoring it. + +## Composition + +- Transform payloads while preserving trace and commit handles with `TracedMessage.map`. This keeps the `Commit` handle attached after parsing. +- When you start child spans for processing, finish them — `TracedMessage.finish` finishes the message's own span; spans you start from `Alma.Tracing` you finish yourself. + +## Integration with Other Libraries + +- Tracing is automatic only when a tracer is active in the host process; the library checks tracer availability and otherwise produces an inactive span. The trace context is propagated as B3 Kafka headers — the consumer extracts it, the producer injects it. No manual header plumbing is required for propagation. +- External checkpoints integrate via `GetCheckpoint = Some f` on the consumer configuration, where `f: GroupId -> TopicPartition -> AsyncResult`. See `examples.md` → External Checkpoint. +- Health checks come from `Checker.defaultChecker` / `IntervalChecker.defaultChecker`; attach them to the configuration's `Checker` / `IntervalChecker` fields to gate consuming/producing on cluster and topic availability with retry and `Alma.Metrics` status marking. + +## Naming Conventions + +- Single-case DU wrappers (`BrokerList of string`, `Offset of int64`, `EventId of Guid`, …) each have a `[]` companion module exposing `value` and constructors. Follow this module-per-type convention for any helper you add. +- Namespace is `Alma.Kafka`; qualify modules (`Consumer.`, `Producer.`, `Admin.`) rather than opening everything. + +## Testing Recommendations + +- Use `GroupId.Random` in tests to force reading a topic from the beginning with an isolated group. +- The repository's own automated coverage is limited to trace-propagation tests; consumer/producer behavior against a real broker is verified manually (the `example/` project ships a `docker-compose.yaml` with Kafka for local runs). Write integration tests against a disposable broker when changing consume/produce logic. diff --git a/.claude/skills b/.claude/skills new file mode 120000 index 0000000..2b7a412 --- /dev/null +++ b/.claude/skills @@ -0,0 +1 @@ +../.agents/skills \ No newline at end of file diff --git a/AGENTS.md b/AGENTS.md index 227d157..24c1955 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,5 +1,7 @@ # AGENTS.md — Alma.Kafka (fkafka) +This repo ships Agent Skill for the `Alma.Kafka` library. Compatible agents discover it automatically; see `.agents/skills/fkafka/SKILL.md`. + ## Project Purpose F# library (`Alma.Kafka`) for producing and consuming messages to/from Apache Kafka streams. Provides a typed, traced API with consumer lag monitoring, health checking, manual/auto commit modes, external checkpoint support, event schema definitions, metadata parsing, and trace propagation via Kafka headers. Published as a NuGet package. diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..c3dccb5 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,2 @@ + +@AGENTS.md