# Configuration Reference (/integrations/watermill/reference/configuration)



This page is the authoritative configuration reference for the `watermill-kubemq` plugin:
every config field with its type and default, the enums and validation rules, the marshaling
contract, reserved metadata keys, and pinned dependency versions. The methods and constructors
themselves are in the [API reference](/integrations/watermill/reference/api). Every entry is drawn from the plugin source;
where the repository README and the source disagree, the source wins and the discrepancy is
called out.

## Package and import [#package-and-import]

The plugin module is published at `github.com/kubemq-io/watermill-kubemq`. All public types
live in the `pkg/kubemq` subpackage, conventionally imported under the alias `kubemq`:

```go
import (
    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill/message"
    kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)
```

Install it with:

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

The plugin is a native gRPC client built on `github.com/kubemq-io/kubemq-go/v2`. It connects
to the KubeMQ broker's gRPC port (`50000`) directly — there is no HTTP connector and no
server-side enable flag to set, unlike the HTTP-based connectors. A broker reachable on
`50000` is all that is required:

<RunKubeMQ ports="[50000, 8080, 9090]" />

Port `50000` is the gRPC API used by this plugin, `8080` exposes the REST API and health
endpoint, and `9090` exposes Prometheus metrics.

## PublisherConfig [#publisherconfig]

Configures `NewPublisher`. `QueueMessagePolicy` is only consulted when `Pattern` is
`PatternQueues`.

<TypeTable
  type="{
  Address: { type: &#x22;string&#x22;, description: &#x22;KubeMQ server host:port. Required unless ExistingClient is set.&#x22; },
  ClientID: { type: &#x22;string&#x22;, description: &#x22;Unique identifier for this publisher. A random UUID is generated if empty.&#x22;, default: &#x22;auto&#x22; },
  AuthToken: { type: &#x22;string&#x22;, description: &#x22;JWT or token for authentication.&#x22;, default: &#x22;\&#x22;\&#x22;&#x22; },
  TLS: { type: &#x22;*TLSConfig&#x22;, description: &#x22;TLS configuration for the connection.&#x22;, default: &#x22;nil&#x22; },
  Pattern: { type: &#x22;PatternType&#x22;, description: &#x22;PatternEvents, PatternEventsStore, or PatternQueues. Required.&#x22; },
  Marshaler: { type: &#x22;Marshaler&#x22;, description: &#x22;Converts Watermill messages to KubeMQ format.&#x22;, default: &#x22;DefaultMarshaler{}&#x22; },
  ExistingClient: { type: &#x22;*kubemqSDK.Client&#x22;, description: &#x22;Reuse an existing kubemq-go client. Caller owns and closes it.&#x22;, default: &#x22;nil&#x22; },
  DisableStreaming: { type: &#x22;bool&#x22;, description: &#x22;Use synchronous sends instead of the streaming APIs.&#x22;, default: &#x22;false&#x22; },
  QueueMessagePolicy: { type: &#x22;*QueueMessagePolicy&#x22;, description: &#x22;Default delivery policy for queue messages (Queues only).&#x22;, default: &#x22;nil&#x22; },
  Logger: { type: &#x22;watermill.LoggerAdapter&#x22;, description: &#x22;Watermill logger adapter.&#x22;, default: &#x22;watermill.NopLogger{}&#x22; },
}"
/>

## SubscriberConfig [#subscriberconfig]

Configures `NewSubscriber`. The `MaxItems` and `WaitTimeoutSeconds` fields apply only to the
Queues pattern; the `EventsStore*` fields apply only to the EventsStore pattern.

<TypeTable
  type="{
  Address: { type: &#x22;string&#x22;, description: &#x22;KubeMQ server host:port. Required unless ExistingClient is set.&#x22; },
  ClientID: { type: &#x22;string&#x22;, description: &#x22;Unique identifier for this subscriber.&#x22;, default: &#x22;auto&#x22; },
  AuthToken: { type: &#x22;string&#x22;, description: &#x22;JWT or token for authentication.&#x22;, default: &#x22;\&#x22;\&#x22;&#x22; },
  TLS: { type: &#x22;*TLSConfig&#x22;, description: &#x22;TLS configuration for the connection.&#x22;, default: &#x22;nil&#x22; },
  Pattern: { type: &#x22;PatternType&#x22;, description: &#x22;PatternEvents, PatternEventsStore, or PatternQueues. Required.&#x22; },
  Unmarshaler: { type: &#x22;Unmarshaler&#x22;, description: &#x22;Converts KubeMQ messages to Watermill format.&#x22;, default: &#x22;DefaultMarshaler{}&#x22; },
  ExistingClient: { type: &#x22;*kubemqSDK.Client&#x22;, description: &#x22;Reuse an existing kubemq-go client.&#x22;, default: &#x22;nil&#x22; },
  ConsumerGroup: { type: &#x22;string&#x22;, description: &#x22;Competing-consumers group name.&#x22;, default: &#x22;\&#x22;\&#x22;&#x22; },
  MaxItems: { type: &#x22;int32&#x22;, description: &#x22;Messages per poll batch (Queues only). Defaults to 1.&#x22;, default: &#x22;1&#x22; },
  WaitTimeoutSeconds: { type: &#x22;int32&#x22;, description: &#x22;Server-side poll timeout in seconds (Queues only). Defaults to 1.&#x22;, default: &#x22;1&#x22; },
  EventsStoreStartOption: { type: &#x22;EventsStoreStartOption&#x22;, description: &#x22;Where to start consuming (EventsStore only).&#x22;, default: &#x22;StartFromNew&#x22; },
  EventsStoreSequence: { type: &#x22;int64&#x22;, description: &#x22;Sequence number for StartFromSequence.&#x22;, default: &#x22;0&#x22; },
  EventsStoreStartTime: { type: &#x22;time.Time&#x22;, description: &#x22;Start time for StartFromTime.&#x22;, default: &#x22;zero&#x22; },
  EventsStoreTimeDelta: { type: &#x22;time.Duration&#x22;, description: &#x22;Duration for StartFromTimeDelta.&#x22;, default: &#x22;0&#x22; },
  Logger: { type: &#x22;watermill.LoggerAdapter&#x22;, description: &#x22;Watermill logger adapter.&#x22;, default: &#x22;watermill.NopLogger{}&#x22; },
}"
/>

## CQConfig [#cqconfig]

Configures `NewCQPublisher`.

<TypeTable
  type="{
  Address: { type: &#x22;string&#x22;, description: &#x22;KubeMQ server host:port. Required unless ExistingClient is set.&#x22; },
  ClientID: { type: &#x22;string&#x22;, description: &#x22;Unique identifier for this CQ publisher.&#x22;, default: &#x22;auto&#x22; },
  AuthToken: { type: &#x22;string&#x22;, description: &#x22;JWT or token for authentication.&#x22;, default: &#x22;\&#x22;\&#x22;&#x22; },
  TLS: { type: &#x22;*TLSConfig&#x22;, description: &#x22;TLS configuration for the connection.&#x22;, default: &#x22;nil&#x22; },
  ExistingClient: { type: &#x22;*kubemqSDK.Client&#x22;, description: &#x22;Reuse an existing kubemq-go client.&#x22;, default: &#x22;nil&#x22; },
  DefaultTimeout: { type: &#x22;time.Duration&#x22;, description: &#x22;Default timeout for commands and queries when a per-call timeout is <= 0.&#x22;, default: &#x22;5s&#x22; },
  CacheKey: { type: &#x22;string&#x22;, description: &#x22;Default cache key applied to every SendQuery call.&#x22;, default: &#x22;\&#x22;\&#x22;&#x22; },
  CacheTTL: { type: &#x22;time.Duration&#x22;, description: &#x22;TTL paired with CacheKey for default query caching.&#x22;, default: &#x22;0&#x22; },
  Marshaler: { type: &#x22;MarshalerUnmarshaler&#x22;, description: &#x22;Marshals requests and unmarshals query replies.&#x22;, default: &#x22;DefaultMarshaler{}&#x22; },
  Logger: { type: &#x22;watermill.LoggerAdapter&#x22;, description: &#x22;Watermill logger adapter.&#x22;, default: &#x22;watermill.NopLogger{}&#x22; },
}"
/>

<Callout type="warn">
  The repository README's `CQConfig` table omits three fields that exist in the source:
  `CacheKey`, `CacheTTL`, and `Marshaler`. They are documented above from
  `pkg/kubemq/config.go`. The README also marks `DefaultTimeout` as "(required)", but the
  source applies a default of `5 * time.Second` when it is `<= 0`, so it is effectively optional.
</Callout>

## QueueMessagePolicy [#queuemessagepolicy]

Default delivery policy for queue messages, set on `PublisherConfig.QueueMessagePolicy` and
used only by the Queues pattern.

<TypeTable
  type="{
  ExpirationSeconds: { type: &#x22;int&#x22;, description: &#x22;Message TTL in seconds (0 = no expiration).&#x22;, default: &#x22;0&#x22; },
  DelaySeconds: { type: &#x22;int&#x22;, description: &#x22;Delay before the message becomes visible to consumers.&#x22;, default: &#x22;0&#x22; },
  MaxReceiveCount: { type: &#x22;int&#x22;, description: &#x22;Max delivery attempts before the message is routed to MaxReceiveQueue.&#x22;, default: &#x22;0&#x22; },
  MaxReceiveQueue: { type: &#x22;string&#x22;, description: &#x22;Dead-letter channel name for messages exceeding MaxReceiveCount.&#x22;, default: &#x22;\&#x22;\&#x22;&#x22; },
}"
/>

<Callout type="info">
  These four fields are typed `int` (and `string` for `MaxReceiveQueue`) in
  `pkg/kubemq/config.go`. The README reference table lists `ExpirationSeconds`, `DelaySeconds`,
  and `MaxReceiveCount` as `int32` — the source type is `int`.
</Callout>

## TLSConfig [#tlsconfig]

TLS settings for the connection, set on the `TLS` field of any config. File-based credentials
(`CertFile`) take precedence over PEM data (`CertData`).

<TypeTable
  type="{
  CertFile: { type: &#x22;string&#x22;, description: &#x22;Path to a CA/cert file; mapped to kubemq-go WithCredentials.&#x22;, default: &#x22;\&#x22;\&#x22;&#x22; },
  CertData: { type: &#x22;string&#x22;, description: &#x22;PEM certificate data; mapped to kubemq-go WithCertificate (used only when CertFile is empty).&#x22;, default: &#x22;\&#x22;\&#x22;&#x22; },
  ServerOverrideDomain: { type: &#x22;string&#x22;, description: &#x22;Server name override for certificate verification.&#x22;, default: &#x22;\&#x22;\&#x22;&#x22; },
  InsecureSkipVerify: { type: &#x22;bool&#x22;, description: &#x22;Skip certificate verification. Do not use in production.&#x22;, default: &#x22;false&#x22; },
}"
/>

## Enums [#enums]

### PatternType [#patterntype]

Selects the KubeMQ messaging pattern for a Publisher or Subscriber. Each instance serves
exactly one pattern.

| Constant             | Value | `String()`     |
| -------------------- | ----- | -------------- |
| `PatternEvents`      | `0`   | `events`       |
| `PatternEventsStore` | `1`   | `events_store` |
| `PatternQueues`      | `2`   | `queues`       |

Any value outside this range fails validation, and `String()` returns `unknown`.

### EventsStoreStartOption [#eventsstorestartoption]

Controls the starting position for an EventsStore subscription
(`SubscriberConfig.EventsStoreStartOption`).

| Constant             | Value | Description                                                     |
| -------------------- | ----- | --------------------------------------------------------------- |
| `StartFromNew`       | `0`   | Only events published after subscribing (default).              |
| `StartFromFirst`     | `1`   | Replay all stored events from the beginning.                    |
| `StartFromLast`      | `2`   | Replay the last event, then continue with new ones.             |
| `StartFromSequence`  | `3`   | Replay from a specific sequence number (`EventsStoreSequence`). |
| `StartFromTime`      | `4`   | Replay from a specific point in time (`EventsStoreStartTime`).  |
| `StartFromTimeDelta` | `5`   | Replay from now minus a duration (`EventsStoreTimeDelta`).      |

## Validation rules [#validation-rules]

Each config exposes a `Validate()` method, invoked automatically by its constructor.
Validation both rejects invalid input and applies defaults in place. The table below
summarizes the enforced rules.

| Config                   | Rule                                                                                                                                                                                                                 |
| ------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| All                      | `Address` is required when `ExistingClient` is `nil` (otherwise an error is returned).                                                                                                                               |
| Publisher / Subscriber   | `Pattern` must be within `PatternEvents`..`PatternQueues`; out-of-range values error.                                                                                                                                |
| Publisher / Subscriber   | A `nil` `Marshaler`/`Unmarshaler` defaults to `DefaultMarshaler{}`; a `nil` `Logger` defaults to `watermill.NopLogger{}`.                                                                                            |
| Subscriber (Queues)      | `MaxItems <= 0` defaults to `1`; `WaitTimeoutSeconds <= 0` defaults to `1`.                                                                                                                                          |
| Subscriber (EventsStore) | `StartFromSequence` requires `EventsStoreSequence > 0`; `StartFromTime` requires a non-zero `EventsStoreStartTime`; `StartFromTimeDelta` requires `EventsStoreTimeDelta > 0`. Each unmet condition returns an error. |
| CQ                       | `DefaultTimeout <= 0` defaults to `5 * time.Second`; a `nil` `Marshaler` defaults to `DefaultMarshaler{}`; a `nil` `Logger` defaults to `watermill.NopLogger{}`.                                                     |

## Marshaling [#marshaling]

The plugin converts between Watermill messages and KubeMQ messages through small intermediate
types and three interfaces.

| Type                   | Kind      | Purpose                                                                                         |
| ---------------------- | --------- | ----------------------------------------------------------------------------------------------- |
| `Marshaler`            | interface | `Marshal(topic string, msg *message.Message) (*MarshaledMessage, error)` — Watermill to KubeMQ. |
| `Unmarshaler`          | interface | `Unmarshal(msg *ReceivedMessage) (*message.Message, error)` — KubeMQ to Watermill.              |
| `MarshalerUnmarshaler` | interface | Embeds both `Marshaler` and `Unmarshaler` (used by `CQConfig.Marshaler`).                       |
| `DefaultMarshaler`     | struct    | Default implementation of `MarshalerUnmarshaler`.                                               |
| `MarshaledMessage`     | struct    | `{ Body []byte; Tags map[string]string }` — outbound intermediate.                              |
| `ReceivedMessage`      | struct    | `{ ID string; Body []byte; Tags map[string]string }` — inbound intermediate.                    |

`DefaultMarshaler` maps the Watermill payload to the KubeMQ message `Body`, and Watermill
metadata to KubeMQ `Tags`. The Watermill message UUID is stored under a reserved tag key
exported as the constant `WatermillUUIDTag`:

```go
const WatermillUUIDTag = "_watermill_uuid"
```

On marshal, the UUID is written to this tag, and supplying `_watermill_uuid` as a metadata key
yourself is rejected with an error (it is reserved). On unmarshal, the UUID is recovered from
the tag; if absent, the KubeMQ message `ID` is used; if that is also empty, a fresh
`watermill.NewUUID()` is generated. The `_watermill_uuid` tag is stripped from the
reconstructed Watermill metadata.

To customize wire format, implement `Marshaler`/`Unmarshaler` (or `MarshalerUnmarshaler` for
CQ) and set it on the corresponding config field — see the
[Connection & Configuration guide](/integrations/watermill/how-to/configuration) for a worked compressed-marshaler example.

## Reserved and auto-populated metadata keys [#reserved-and-auto-populated-metadata-keys]

Several metadata keys carry special meaning. `_watermill_uuid` is reserved (you cannot set
it). The `_kubemq_*` keys are set by the subscriber on inbound messages from KubeMQ message
attributes. The W3C Trace Context keys are injected on publish and extracted on receive by the
OpenTelemetry propagator.

| Key                 | Direction | Set by                           | Notes                                                                  |
| ------------------- | --------- | -------------------------------- | ---------------------------------------------------------------------- |
| `_watermill_uuid`   | both      | `DefaultMarshaler`               | Stores the Watermill UUID. Reserved — supplying it as metadata errors. |
| `_kubemq_channel`   | inbound   | Subscriber (Events, EventsStore) | KubeMQ channel the message arrived on.                                 |
| `_kubemq_sequence`  | inbound   | Subscriber (EventsStore)         | EventsStore sequence number, as a string.                              |
| `_kubemq_timestamp` | inbound   | Subscriber (EventsStore)         | EventsStore timestamp, RFC3339Nano.                                    |
| `traceparent`       | both      | OTel propagator                  | W3C Trace Context — injected on publish, extracted on receive.         |
| `tracestate`        | both      | OTel propagator                  | W3C Trace Context vendor state.                                        |

<Callout type="info">
  Trace context propagation uses the globally configured OpenTelemetry `TextMapPropagator`. If no
  propagator is registered in your process, the `traceparent`/`tracestate` keys are not added.
  See the [Middleware & Observability guide](/integrations/watermill/how-to/middleware-and-observability) for the
  OpenTelemetry setup.
</Callout>

## Dependency versions [#dependency-versions]

Pinned core dependencies, from `go.mod`:

| Dependency                           | Version   |
| ------------------------------------ | --------- |
| Go (toolchain)                       | `1.25.0`  |
| `github.com/ThreeDotsLabs/watermill` | `v1.5.1`  |
| `github.com/kubemq-io/kubemq-go/v2`  | `v2.0.3`  |
| `go.opentelemetry.io/otel`           | `v1.44.0` |

## Related [#related]

<Cards>
  <Card title="API Reference" href="/integrations/watermill/reference/api" description="Constructors and the Publisher, Subscriber, and CQPublisher method signatures." />

  <Card title="Connection & Configuration" href="/integrations/watermill/how-to/configuration" description="Worked examples for address, auth, TLS, client reuse, and health checks." />
</Cards>