# Watermill (/integrations/watermill)



[`kubemq-watermill`](https://github.com/kubemq-io/kubemq-watermill) is a production-ready
[Watermill](https://watermill.io/) pub/sub plugin for KubeMQ. It implements Watermill's
`message.Publisher` and `message.Subscriber` interfaces across three KubeMQ messaging
patterns — Events, EventsStore, and Queues — plus a native `CQPublisher` for Commands and
Queries (request-reply). Drop it into an existing Watermill application and your Router,
handlers, and middleware run unchanged on top of KubeMQ.

It is a **native gRPC client** built on the `kubemq-go/v2` SDK — not an HTTP connector.
If you need protocol bridging instead (REST, MCP, A2A, CloudEvents), see the
[Connectors](/connectors) overview. For the messaging model behind the patterns,
see the core docs: [Events](/learn/events), [Events Store](/learn/events-store),
[Queues](/learn/queues), and [RPC](/learn/rpc).

## Why Watermill on KubeMQ [#why-watermill-on-kubemq]

* **Three messaging patterns** — Events (fire-and-forget), EventsStore (persistent with
  replay), and Queues (reliable with explicit ack/nack), each behind the same Watermill
  `Publisher`/`Subscriber` interfaces.
* **Native CQPublisher** — wraps KubeMQ's Commands and Queries APIs directly for
  low-latency request-reply, with execution confirmation and query caching.
* **Full Watermill compatibility** — works with the Watermill Router and the entire
  standard middleware stack (Retry, Throttle, CorrelationID, Poison Queue, CircuitBreaker).
* **Ack/Nack bridged to KubeMQ** — Watermill `msg.Ack()`/`msg.Nack()` map to KubeMQ queue
  settlement; a Nack returns the message to the queue for redelivery.
* **Built-in DLQ, OTel, and metrics** — native dead-lettering via `QueueMessagePolicy`,
  W3C Trace Context propagation through KubeMQ Tags, and Watermill's Prometheus component.
* **Verified compatibility** — passes the full Watermill `pubsub/tests.TestPubSub` suite.

## Architecture [#architecture]

Your Watermill Router, Publisher, and Subscriber call the plugin, which uses
`kubemq-go/v2` to open a gRPC connection to the KubeMQ broker on port `50000`. There is no
HTTP connector and no server-side translation layer — the broker's gRPC server is always
on, so the plugin works as soon as a broker is reachable.

<Mermaid
  chart="`
graph LR
ROUTER[&#x22;Watermill app<br/>Router · Publisher · Subscriber&#x22;]
CQ[&#x22;CQPublisher<br/>(Commands / Queries)&#x22;]
PLUGIN{{&#x22;watermill-kubemq<br/>plugin&#x22;}}
SDK[&#x22;kubemq-go/v2<br/>gRPC client&#x22;]
BROKER[&#x22;KubeMQ Broker&#x22;]

ROUTER -- &#x22;message.Publisher / Subscriber&#x22; --> PLUGIN
CQ -- &#x22;SendCommand / SendQuery&#x22; --> PLUGIN
PLUGIN -- &#x22;NewClient&#x22; --> SDK
SDK -- &#x22;gRPC :50000&#x22; --> BROKER

class ROUTER,CQ client
class PLUGIN aiway
class SDK,BROKER broker
`"
/>

*The plugin maps Watermill's Publisher/Subscriber model onto KubeMQ patterns over the native gRPC SDK on port 50000.*

## Capabilities [#capabilities]

Each Publisher and Subscriber serves exactly one KubeMQ pattern, selected with the
`Pattern` field. Commands and Queries use the standalone `CQPublisher`.

<Cards>
  <Card title="Events" href="/integrations/watermill/how-to/events" description="Fire-and-forget pub/sub with PatternEvents — fan-out, consumer groups, and streaming publish." />

  <Card title="Events Store" href="/integrations/watermill/how-to/events-store" description="Persistent, replayable events with PatternEventsStore — start from first, last, a sequence, or a point in time." />

  <Card title="Queues" href="/integrations/watermill/how-to/queues" description="Reliable point-to-point delivery with explicit ack/nack, batching, and native DLQ routing." />

  <Card title="Commands & Queries" href="/integrations/watermill/how-to/commands-queries" description="Native CQPublisher for low-latency request-reply over KubeMQ Commands and Queries." />
</Cards>

## Pattern selection [#pattern-selection]

KubeMQ supports three messaging patterns. Each Publisher/Subscriber instance serves
exactly one — pick the one that matches your delivery and persistence needs:

| Pattern         | Delivery      | Ack/Nack             | Persistence                 | Best for                                       |
| --------------- | ------------- | -------------------- | --------------------------- | ---------------------------------------------- |
| **Events**      | At-most-once  | No (fire-and-forget) | No                          | Real-time notifications, metrics, logs         |
| **EventsStore** | At-least-once | Offset auto-advance  | Yes (replay from any point) | Event sourcing, audit trails, stream replay    |
| **Queues**      | At-least-once | Explicit ack/nack    | Yes (until acked)           | Task queues, job processing, reliable delivery |

For request-reply, reach for the native `CQPublisher` (single hop, lower latency) or
Watermill's `requestreply` component over Queues (middleware-compatible) — see
[Commands & Queries](/integrations/watermill/how-to/commands-queries).

## Supported runtime [#supported-runtime]

| Requirement                          | Version                                       |
| ------------------------------------ | --------------------------------------------- |
| Go                                   | 1.25+                                         |
| `github.com/ThreeDotsLabs/watermill` | v1.5.1                                        |
| `github.com/kubemq-io/kubemq-go/v2`  | v2.0.3                                        |
| KubeMQ broker                        | gRPC on `:50000` (always on — no enable flag) |

Install the plugin with `go get`:

```bash
go get github.com/kubemq-io/watermill-kubemq
```

## Next steps [#next-steps]

<Cards>
  <Card title="Getting Started" href="/integrations/watermill/tutorials/getting-started" description="Run a broker and publish your first Watermill message over KubeMQ in a few minutes." />

  <Card title="Concepts" href="/integrations/watermill/concepts/concepts" description="How the plugin maps Publishers, Subscribers, and the Router onto KubeMQ's patterns, marshaling, and ack semantics." />

  <Card title="Guides" href="/integrations/watermill/how-to/configuration" description="Connection and TLS configuration, middleware, OpenTelemetry, Prometheus, and KEDA autoscaling." />

  <Card title="Reference" href="/integrations/watermill/reference/configuration" description="Full PublisherConfig, SubscriberConfig, CQConfig, QueueMessagePolicy, and API reference." />
</Cards>

New to KubeMQ? Start with the [What is an integration?](/integrations#what-an-integration-is)
mental model, then the [KubeMQ Getting Started guide](/deploy).