Twin Kafka (dual-cluster bridge)¶

Guide: connect an application to two Kafka clusters and bridge messages between them with the opt-in twin-kafka library.

At a glance

What - twin-kafka adds a SECOND Kafka cluster to an application built on minimalist-kafka: a secondary notification function (secondary.kafka.notification), an optional secondary flow adapter (yaml.secondary.kafka.flow.adapter), and an optional secondary Schema Registry - each the exact counterpart of its primary sibling.

Why a separate module - dual-cluster connectivity is a special case (typically a bridge between an on-prem and a cloud cluster). Keeping it out of minimalist-kafka means every ordinary single-cluster application stays on the lightweight library; only the bridge application adds this jar.

A bridge is just a flow - consume from one cluster's adapter, publish through the other cluster's notification function. Trace continuity and business correlation-id propagation across both Kafka hops come from the existing machinery.

For developers building a bridge (or any app that must speak to two Kafka clusters).

How it works¶

twin-kafka depends on minimalist-kafka and reuses its machinery through explicit extension seams - the flow adapter, publisher, retry/DLQ handling, templates mechanics, and Confluent codec are the same classes, instantiated a second time against the second cluster. The two runtimes are fully separate: dead letters from a secondary binding are written to the secondary cluster, and each cluster's schema-id cache is isolated (Confluent global ids are only unique within one registry).

Concern	Primary cluster (minimalist-kafka)	Secondary cluster (twin-kafka)
Publish function	`simple.kafka.notification`	`secondary.kafka.notification`
Inbound bindings	`yaml.kafka.flow.adapter`	`yaml.secondary.kafka.flow.adapter`
Producer template	`kafka-producer.properties`	`secondary-kafka-producer.properties`
Consumer template	`kafka-consumer.properties`	`secondary-kafka-consumer.properties`
Schema Registry (optional)	`schema.registry.url` + `schema-registry.properties`	`secondary.schema.registry.url` + `secondary-schema-registry.properties`
Client opt-out (default `true`)	`kafka.producer.enabled` / `kafka.consumer.enabled`	`secondary.kafka.producer.enabled` / `secondary.kafka.consumer.enabled`
Outbound header names	`kafka.correlation.id.header` / `kafka.trace.id.header` / `kafka.traceparent.header`	`secondary.kafka.correlation.id.header` / `secondary.kafka.trace.id.header` / `secondary.kafka.traceparent.header` (fall back to the globals)

All secondary templates follow the same mechanics as the primary ones: loaded from the bundled classpath template by default, ${ENV_VAR:default} substitution, and OAuth token endpoint URLs auto-registered on the JVM allow-list. That includes the consumer rebalance protocol: each cluster resolves group.protocol=auto independently, so the primary may run the KIP-848 consumer protocol while the secondary stays classic (or vice versa). Externalization of configuration is opt-in - point the location key at a file rendered by the devops pipeline (e.g. file:/tmp/config/...), optionally with a classpath fallback as a comma-separated list. The default bootstrap is ${SECONDARY_KAFKA_BOOTSTRAP_SERVERS:127.0.0.1:8092} - matching the local dual-broker helper below.

The bridge pattern¶

Bridge on-prem (primary) to cloud (secondary) with one flow per direction - no bridge code, just flow YAML. Inbound bindings on the source cluster:

# kafka-flow-adapter.yaml (primary = on-prem)
consumer:
  - topic: 'orders.onprem'
    flow: 'bridge-to-cloud'
    group: 'bridge-up-group'

The bridge flow republishes through the other cluster's notification function, carrying the business correlation-id (model.cid, seeded by the adapter) and stamping a fresh W3C traceparent:

# bridge-to-cloud.yml
first.task: 'secondary.kafka.notification'
tasks:
  - input:
      - 'input.body -> *'
      - 'text(orders.cloud) -> header.topic'
      - 'model.cid -> header.cid'
    process: 'secondary.kafka.notification'
    output: []
    description: 'Republish the on-prem message to the cloud cluster'
    execution: end

The reverse direction mirrors it: a binding in yaml.secondary.kafka.flow.adapter routes the cloud topic into a flow that publishes through simple.kafka.notification. Because each hop is a normal flow, the bridge is observable (continuous trace across both clusters), governable (per-binding retry + dead-letter on the originating cluster), and extensible (enrich/filter/transform tasks can sit between consume and republish).

A one-way bridge: switch off the clients you do not use¶

A bridge that only flows on-prem → cloud consumes on the primary and produces on the secondary — so two of the four clients are dead weight. That matters beyond tidiness: a managed cluster issues credentials per client, and a Confluent console that was never asked for a primary producer API key has none to give. Building the unused client then fails the deployment. Declare the direction instead:

# on-prem -> cloud, one way. Two clusters, one client each.
kafka.consumer.enabled=true
kafka.producer.enabled=false
secondary.kafka.consumer.enabled=false
secondary.kafka.producer.enabled=true

yaml.kafka.flow.adapter=classpath:/kafka-flow-adapter.yaml
# no yaml.secondary.kafka.flow.adapter - nothing consumes on the cloud side

Each cluster reads only its own keys, and each diagnoses itself: publishing through simple.kafka.notification here fails naming kafka.producer.enabled, and the secondary counterpart names secondary.kafka.producer.enabled. Two consequences to plan for:

Dead letters need the producer of the cluster they came from. The primary binding above consumes from on-prem, so its dlq-topic would be written to on-prem — through the producer this configuration just disabled. That contradiction fails startup rather than dropping poison messages silently. Either keep kafka.producer.enabled=true (a DLQ-only producer credential) or handle failures without a DLQ.
Health checks still cover both legs. kafka.health probes through the consumer template as usual; secondary.kafka.health finds its consumer disabled and probes through the secondary producer template instead, so mandatory.health.dependencies=kafka.health, secondary.kafka.health keeps working unchanged. See health check.

Asymmetric Schema Registry¶

The Schema Registry is optional and per-cluster. A common real-world topology pairs an on-prem Apache Kafka (no registry, raw byte[]) with a cloud Confluent cluster (with a registry): leave schema.registry.url unset and set only secondary.schema.registry.url. Subject-driven Confluent serialization (subject header on secondary.kafka.notification) and schema.enabled decoding on secondary bindings then work against the secondary registry only, while the primary side keeps the minimalist raw-byte[] behavior. Each side fails fast with its own config key named in the error if a subject is used where no registry is configured.

Per-direction identity on the secondary cluster. When the secondary registry grants access per direction (most visibly CSFLE key access through separate produce and consume identity pools), set secondary.schema.registry.consumer.properties to give the secondary flow adapter its own codec, built under the secondary.schema.registry.consumer prefix against the same secondary.schema.registry.url — the exact twin of minimalist-kafka's consumer-side identity, with the same rules: presence is the opt-in (blank stays off), secondary.schema.registry.consumer.serde.* overrides layer on the template, and the consumer codec reads only its own prefix.

Deployment topologies¶

Because connection/security is entirely template-driven and the registry is independently optional per cluster, common dual-cluster pairings are configuration-only:

Topology	Primary	Secondary	Configuration notes
Apache + Confluent	on-prem Apache Kafka (no registry)	Confluent Kafka with registry	The asymmetric topology above - set only `secondary.schema.registry.url`. This exact pairing is proven end-to-end by the module's test suite.
Confluent + Confluent	Confluent with registry 1	Confluent with registry 2	Set both registry URLs. The two registries are fully independent - separate auth templates (even two different OAuth identity providers) and separate schema-id caches. See the id-translation note below.
Apache + Azure Event Hubs	on-prem Apache Kafka	Event Hubs Kafka endpoint	Kafka-protocol compatible via the secondary templates: `SASL_SSL` with `PLAIN` (`$ConnectionString`) or `OAUTHBEARER` (Entra ID - the token URL is auto-allow-listed). Leave `secondary.schema.registry.url` unset (Azure Schema Registry is not Confluent-protocol compatible) and pre-provision the event hubs - Event Hubs does not honor Kafka topic auto-creation. Consumer groups, manual commit-after-process, and record headers (which carry `traceparent` and the correlation-id) are all supported by the Event Hubs Kafka endpoint.

The design is symmetric: each pairing also works with the roles swapped - "primary" is simply the cluster configured through minimalist-kafka's base templates.

Bridging framed payloads between two registries¶

Confluent global schema ids are only unique within one registry, and the bridge does not translate ids. To bridge a Confluent-framed message between two registries, let the flow decode-and-re-encode: schema.enabled: true on the inbound binding (decode against the source registry) and a subject header on the outbound notification (re-frame against the target registry's own id for that subject). Do not relay the raw framed bytes without schema.enabled - the embedded id would be meaningless on the other registry.

Local development: one helper, two brokers¶

The kafka-standalone helper can emulate the dual-cluster topology in one process - set dual.servers=true (in its application.properties or as -Ddual.servers=true) and it starts broker 1 on 127.0.0.1:9092 and broker 2 on 127.0.0.1:8092 with separate log directories:

Note: x.y.z denotes the current Mercury version shown in the root pom.xml.

cd helpers/kafka-standalone && java -Ddual.servers=true -jar target/kafka-standalone-x.y.z-exec.jar

The shipped templates' defaults (127.0.0.1:9092 primary, 127.0.0.1:8092 secondary) line up with this, so a bridge application runs locally with zero connection configuration.

For a complete runnable walkthrough — an HTTP edge, a pure-YAML bridge, and a system-of-record crossing the two emulated clusters with trace/correlation continuity — see the twin-kafka-demo worked example.

Configuration keys¶

Key	Default	Description
`yaml.secondary.kafka.flow.adapter`	-	Secondary adapter config location; unset = secondary inbound adapter off.
`secondary.kafka.producer.enabled`	`true`	Set `false` when the secondary cluster issues no producer credentials — see one-way bridge. A secondary binding with `dlq-topic` then fails startup.
`secondary.kafka.consumer.enabled`	`true`	Set `false` when the secondary cluster issues no consumer credentials; `secondary.kafka.health` then probes through the secondary producer template.
`secondary.kafka.producer.properties`	`classpath:/secondary-kafka-producer.properties`	Secondary producer template location. Set to an external file path (or explicit fallback list) to externalize.
`secondary.kafka.consumer.properties`	`classpath:/secondary-kafka-consumer.properties`	Secondary consumer template location. Set to an external file path (or explicit fallback list) to externalize.
`secondary.schema.registry.url`	-	Secondary cluster's Schema Registry URL; unset = schema features off on that cluster.
`secondary.schema.registry.properties`	`classpath:/secondary-schema-registry.properties`	Secondary registry client template - auth/SSL passed verbatim to the Confluent client. Set to an external file path (or explicit fallback list) to externalize.
`secondary.schema.registry.cache.ttl`	`30m`	Secondary schema cache TTL (the cache is separate from the primary's - schema ids are per-registry).
`secondary.schema.registry.consumer.properties`	-	Opt in to a separate registry identity for the secondary consume side — the twin of `schema.registry.consumer.properties`: the secondary adapter's own registry client template (the secondary producer's file or a second one), with `secondary.schema.registry.consumer.serde.*` overrides on top. Unset or blank = the secondary adapter shares the secondary producer's codec.
`secondary.kafka.health.timeout`	falls back to `kafka.health.timeout` (`5s`)	Probe timeout for `secondary.kafka.health`.
`secondary.kafka.health.startup.grace`	falls back to `kafka.health.startup.grace` (`30s`)	Start-up grace for `secondary.kafka.health` (placeholder healthy status while the client warms up).
`secondary.kafka.correlation.id.header`	falls back to `kafka.correlation.id.header` (`cid`)	Outbound correlation-id header on the secondary cluster.
`secondary.kafka.trace.id.header`	falls back to `kafka.trace.id.header` (unset)	Optional outbound trace-id header on the secondary cluster.
`secondary.kafka.traceparent.header`	falls back to `kafka.traceparent.header` (`traceparent`)	Header name carrying the full W3C trace context on the secondary cluster — for backward compatibility with a legacy convention only (prefer the standard `traceparent`); when customized, the notification stamps the same value under both this and the standard name.

The retry/dead-letter tuning keys (kafka.dlq.timeout.ms, kafka.flow.max.retries, kafka.flow.retry.backoff.ms) are application-level policy shared by both adapters. Per-binding options in the secondary adapter YAML are identical to the primary's - including topic-pattern, schema.enabled, dlq-topic, second-level routing (flows with flow:// and task:// targets, serializer: 'json', ttl), and the trace.id.header / correlation.id.header / traceparent.header impedance-matching overrides. Dead letters from a secondary binding - whichever target kind failed - are published through the secondary cluster's own publisher. Likewise on the outbound side, secondary.kafka.notification inherits the primary publish function's body handling: byte[] verbatim, or a Map/List auto-serialized to JSON bytes on non-schema publishes.

Health check¶

The module ships secondary.kafka.health - the twin of minimalist-kafka's kafka.health, probing the secondary cluster through the secondary consumer template with the same no-ACL Metadata request, start-up grace, and HTTP 503 semantics. A bridge is only healthy when both clusters are reachable, so a dual-cluster application lists both dependencies:

mandatory.health.dependencies=kafka.health, secondary.kafka.health

The /health dependency list distinguishes the two by service name (kafka vs secondary.kafka, each with its cluster's bootstrap.servers as href). An application that treats the secondary cluster as best-effort can move secondary.kafka.health to optional.health.dependencies instead.

On a one-way bridge the probe follows whichever client that cluster configured: with secondary.kafka.consumer.enabled=false there is no consumer credential to probe with, so secondary.kafka.health builds its probe from secondary-kafka-producer.properties instead — same Metadata request, same no-ACL requirement, same 503 semantics. Both legs of a one-way bridge therefore stay health-checkable with the dependency list unchanged.