> ## Documentation Index
> Fetch the complete documentation index at: https://docs.conduktor.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Conduktor Gateway multi-cluster setup

> Configure Conduktor Gateway to proxy multiple Kafka clusters from a single endpoint. Route clients to topics across clusters using topic aliases and access controls.

Gateway can be configured to communicate with multiple Kafka clusters and expose their topics to your partners. You can:

* direct partners to a **single endpoint**,
* provide them with **access to topics in multiple Kafka clusters** and
* expose topics using **aliases** that can be different from actual topic names.

To set up Gateway to support multi-clusters, you should:

1. Configure **one** main cluster which will be used by Gateway to store its internal state.
2. Set up any number of upstream physical Kafka clusters that you want to expose through Gateway.

<Warning>
  If you're using partner virtual clusters to share data with external third parties, be aware that cluster IDs (e.g., `clusterA`, `clusterB`) may appear in the bootstrap server address or client logs.

  To prevent unintended exposure, **avoid using sensitive names/information in cluster IDs**.
</Warning>

## Choose a routing strategy

For production deployments with multiple clusters, use [SNI routing](/guide/tutorials/sni-routing) on your listeners. With port routing, Gateway assigns a fixed port to every broker across all clusters, so adding or removing a cluster shifts those port assignments and can break existing client connections. SNI routing instead distinguishes brokers by hostname, so adding or removing a cluster leaves other clients unaffected.

Port routing remains a simpler choice for development and test environments — or anywhere you can update client bootstrap servers and restart clients when the set of clusters changes. [Configure listeners](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/listeners).

## Configure the clusters

Define your main and upstream clusters using either a configuration file or environment variables.

<Tabs>
  <Tab title="Use a configuration file">
    Specify your main and upstream cluster configurations, along with a `gateway.roles` entry to mark the upstream clusters.

    ```yaml title="cluster-config.yaml" theme={null}
    config:
      main:
        bootstrap.servers: '<main_bootstrap_servers>:9092'
        security.protocol: 'SASL_SSL'
        sasl.mechanism: 'PLAIN'
        sasl.jaas.config: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="<main_api_key>" password="<main_api_secret>";'
      clusterA:
        bootstrap.servers: '<clusterA_bootstrap_servers>:9092'
        security.protocol: 'SASL_SSL'
        sasl.mechanism: 'PLAIN'
        sasl.jaas.config: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="<clusterA_api_key>" password="<clusterA_api_secret>";'
        gateway.roles: 'upstream' # Note: may be omitted as upstream is the default (used to differentiate from failover clusters)
      clusterB:
        bootstrap.servers: '<clusterB_bootstrap_servers>:9092'
        security.protocol: 'SASL_SSL'
        sasl.mechanism: 'PLAIN'
        sasl.jaas.config: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="<clusterB_api_key>" password="<clusterB_api_secret>";'
        gateway.roles: 'upstream' # Note: may be omitted as upstream is the default (used to differentiate from failover clusters)
    ```

    Then, mount the cluster config file in the Gateway container using the configuration `GATEWAY_BACKEND_KAFKA_SELECTOR`:

    ```yaml theme={null}
    GATEWAY_BACKEND_KAFKA_SELECTOR: 'file : { path: /cluster-config.yaml}'
    ```
  </Tab>

  <Tab title="Use environment variables">
    Configure your main and upstream clusters through environment variables, defined in the Gateway container:

    ```yaml theme={null}
    KAFKA_MAIN_BOOTSTRAP_SERVERS: '<main_bootstrap_servers>:9092'
    KAFKA_MAIN_SECURITY_PROTOCOL: 'SASL_SSL'
    KAFKA_MAIN_SASL_MECHANISM: 'PLAIN'
    KAFKA_MAIN_SASL_JAAS_CONFIG: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="<main_api_key>" password="<main_api_secret>";'

    KAFKA_CLUSTERA_BOOTSTRAP_SERVERS: '<clusterA_bootstrap_servers>:9092'
    KAFKA_CLUSTERA_SECURITY_PROTOCOL: 'SASL_SSL'
    KAFKA_CLUSTERA_SASL_MECHANISM: 'PLAIN'
    KAFKA_CLUSTERA_SASL_JAAS_CONFIG: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="<clusterA_api_key>" password="<clusterA_api_secret>";'
    KAFKA_CLUSTERA_GATEWAY_ROLES: 'upstream' # Note: may be omitted as upstream is the default (used to differentiate from failover clusters)

    KAFKA_CLUSTERB_BOOTSTRAP_SERVERS: '<clusterB_bootstrap_servers>:9092'
    KAFKA_CLUSTERB_SECURITY_PROTOCOL: 'SASL_SSL'
    KAFKA_CLUSTERB_SASL_MECHANISM: 'PLAIN'
    KAFKA_CLUSTERB_SASL_JAAS_CONFIG: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="<clusterB_api_key>" password="<clusterB_api_secret>";'
    KAFKA_CLUSTERB_GATEWAY_ROLES: 'upstream' # Note: may be omitted as upstream is the default (used to differentiate from failover clusters)
    ```
  </Tab>
</Tabs>

## Partner virtual clusters

These steps are for setting up a partner virtual cluster manually.

Alternatively, to simplify the process, we recommend creating a [Partner Zone](/guide/use-cases/third-party-data#create-a-partner-zone) which supports multi-clusters.

You can also [Check out the tutorial on creating Partner Zones with multi-cluster Gateway](/guide/tutorials/partner-zone-multi-cluster).

### 1. Create a partner virtual cluster

First, create a new partner virtual cluster.

<Info>
  For partner virtual clusters, `aclEnabled` has to be `true` and `superUsers` must not be empty.
</Info>

<Tabs>
  <Tab title="CLI">
    Create this YAML file:

    ```yaml title="mypartner.yaml" theme={null}
    ---
    kind: VirtualCluster
    apiVersion: gateway/v2
    metadata:
      name: mypartner
    spec:
      aclEnabled: true
      superUsers:
        - super-user
      type: Partner
    ```

    Then, apply it:

    ```shell theme={null}
    conduktor apply -f mypartner.yaml
    ```
  </Tab>

  <Tab title="API">
    ```sh theme={null}
    curl \
      --request PUT \
      --url 'http://localhost:8888/gateway/v2/virtual-cluster' \
      --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
      --header 'Content-Type: application/json' \
      --data-raw '{
        "kind": "VirtualCluster",
        "apiVersion": "gateway/v2",
        "metadata": {
          "name": "mypartner"
        },
        "spec": {
          "aclEnabled": true,
          "superUsers": [ "super-user" ],
          "type": "Partner"
        }
      }'

    ```
  </Tab>
</Tabs>

### 2. Alias your topics

Finally, create aliases for existing topics in the partner virtual cluster.

<Warning>
  Alias topics within a partner virtual cluster **can only point to topics from the same physical cluster**.
</Warning>

<Tabs>
  <Tab title="CLI">
    Create this YAML file:

    ```yaml title="alias-topics.yaml" theme={null}
    ---
    kind: AliasTopic
    apiVersion: gateway/v2
    metadata:
      name: topic1
      vCluster: mypartner
    spec:
      physicalName: internal-topic-name1
      physicalCluster: clusterA
    ---
    kind: AliasTopic
    apiVersion: gateway/v2
    metadata:
      name: topic2
      vCluster: mypartner
    spec:
      physicalName: internal-topic-name2
      physicalCluster: clusterA
    ```

    Then, apply it:

    ```sh theme={null}
    conduktor apply -f alias-topics.yaml
    ```
  </Tab>

  <Tab title="API">
    ```sh theme={null}
    curl \
      --request PUT \
      --url 'http://localhost:8888/gateway/v2/alias-topic' \
      --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
      --header 'Content-Type: application/json' \
      --data-raw '{
        "kind": "AliasTopic",
        "apiVersion": "gateway/v2",
        "metadata": {
          "name": "topic1",
          "vCluster": "mypartner"
        },
        "spec": {
          "physicalName": "internal-topic-name1",
          "physicalCluster": "clusterA"
        }
      }'

    curl \
      --request PUT \
      --url 'http://localhost:8888/gateway/v2/alias-topic' \
      --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
      --header 'Content-Type: application/json' \
      --data-raw '{
        "kind": "AliasTopic",
        "apiVersion": "gateway/v2",
        "metadata": {
          "name": "topic2",
          "vCluster": "mypartner"
        },
        "spec": {
          "physicalName": "internal-topic-name2",
          "physicalCluster": "clusterA"
        }
      }'
    ```
  </Tab>
</Tabs>

### 3. Create service accounts

Once the virtual cluster is created and contains the topics to expose to your partners, you'll need to create service accounts and configure ACLs (Access Control Lists).

Create two service accounts for the partner virtual cluster: **one super user and one partner user**.

The super user will manage ACLs and grant permissions to the partner user, who will use their account to access the exposed topics.

<Tabs>
  <Tab title="CLI">
    ```yaml title="service-accounts.yaml" theme={null}
    ---
    kind: GatewayServiceAccount
    apiVersion: gateway/v2
    metadata:
      name: super-user
      vCluster: mypartner
    spec:
      type: LOCAL
    ---
    kind: GatewayServiceAccount
    apiVersion: gateway/v2
    metadata:
      name: partner-user
      vCluster: mypartner
    spec:
      type: LOCAL
    ```

    Then, apply it:

    ```shell theme={null}
    conduktor apply -f service-accounts.yaml
    ```

    In order to connect to Gateway using these service accounts, you need to get their associated password.

    ```shell theme={null}
    conduktor run generateServiceAccountToken \
      --username super-user \
      --v-cluster mypartner \
      --life-time-seconds 100000000

    conduktor run generateServiceAccountToken \
      --username partner-user \
      --v-cluster mypartner \
      --life-time-seconds 100000000
    ```
  </Tab>

  <Tab title="API">
    ```sh theme={null}
    curl \
      --request PUT \
      --url 'http://localhost:8888/gateway/v2/service-account' \
      --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
      --header 'Content-Type: application/json' \
      --data-raw '{
        "kind": "GatewayServiceAccount",
        "apiVersion": "gateway/v2",
        "metadata": {
          "name": "super-user",
          "vCluster": "mypartner"
        },
        "spec": {
          "type": "LOCAL"
        }
      }'

    curl \
      --request PUT \
      --url 'http://localhost:8888/gateway/v2/service-account' \
      --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
      --header 'Content-Type: application/json' \
      --data-raw '{
        "kind": "GatewayServiceAccount",
        "apiVersion": "gateway/v2",
        "metadata": {
          "name": "partner-user",
          "vCluster": "mypartner"
        },
        "spec": {
          "type": "LOCAL"
        }
      }'
    ```

    To connect to Gateway using these service accounts, you need to get the associated password.

    ```sh theme={null}
    curl \
      --request POST \
      --url 'http://localhost:8888/gateway/v2/token' \
      --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
      --header 'Content-Type: application/json' \
      --data-raw '{
        "username": "super-user",
        "vCluster": "mypartner",
        "lifeTimeSeconds": 3600000
      }'

    curl \
      --request POST \
      --url 'http://localhost:8888/gateway/v2/token' \
      --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
      --header 'Content-Type: application/json' \
      --data-raw '{
        "username": "partner-user",
        "vCluster": "mypartner",
        "lifeTimeSeconds": 3600000
      }'
    ```
  </Tab>
</Tabs>

Put the admin credentials in a file called `mypartner-super-user.properties`:

```properties title="mypartner-super-user.properties" theme={null}
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="super-user" password="$SUPER-USER-PASSWORD";
```

And the partner credentials in a file called `mypartner-partner-user.properties`:

```properties title="mypartner-partner-user.properties" theme={null}
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="partner-user" password="$PARTNER-USER-PASSWORD";
```

### 4. Create ACLs for the service accounts

Before creating ACLs, you need to know how to reach this partner virtual cluster.

For that, make the following request:

```sh theme={null}
curl \
  --silent \
  --user "admin:conduktor" \
  "http://localhost:8888/gateway/v2/virtual-cluster/mypartner"
```

This will return something like this, with the bootstrap address and client properties:

```json theme={null}
{
  "kind": "VirtualCluster",
  "apiVersion": "gateway/v2",
  "metadata": {
    "name": "mypartner"
  },
  "spec": {
    "aclEnabled": true,
    "superUsers": [ "super-user" ],
    "type": "Partner",
    "bootstrapServers": "<partner_virtual_cluster_bootstrap_address>",
    "clientProperties": {
      "security.protocol": "SASL_PLAINTEXT",
      "sasl.mechanism": "PLAIN",
      "sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username={{username}} password={{password}};"
    }
  }
}
```

From there, you have everything you need to create ACLs for the partner service accounts:

```sh theme={null}
# The partner can consume all the topics of this partner virtual cluster
kafka-acls --bootstrap-server <partner_virtual_cluster_bootstrap_address> \
  --command-config mypartner-super-user.properties \
  --add \
  --allow-principal User:partner-user \
  --consumer \
  --topic "*" \
  --group partner-app

# The partner can produce and consume from the topic1 alias topic
kafka-acls --bootstrap-server <partner_virtual_cluster_bootstrap_address> \
  --command-config mypartner-super-user.properties \
  --add \
  --allow-principal User:partner-user \
  --producer \
  --topic topic1
```

### 5. Test partner virtual cluster access

Now that the partner user has the correct ACLs, you can use their credentials to interact with the alias topics and verify that the permissions are correctly set.

```sh theme={null}
kafka-console-producer --bootstrap-server localhost:6974 \
  --topic topic1 \
  --producer.config mypartner-partner-user.properties

kafka-console-consumer --bootstrap-server localhost:6974 \
  --topic topic2 \
  --consumer.config mypartner-partner-user.properties \
  --group partner-app \
  --from-beginning
```

Once confirmed, simply share the `mypartner-partner-user.properties` file and the correct bootstrap server details with your partner.

## Related resources

* [Share data with third parties](/guide/use-cases/third-party-data)
* [Create Partner Zones with multi-cluster Gateway](/guide/tutorials/partner-zone-multi-cluster)
* [Connect Gateway to Kafka](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/connect-to-kafka)
* [Decide on Virtual Clusters](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/virtual-clusters)
* [Manage service accounts](/guide/tutorials/manage-gateway-service-accounts)
* [Resource reference](/guide/reference/gateway-reference)