Confluent Cloud

This guide goes through the required steps to connect Materialize to a Confluent Cloud Kafka cluster.

💡 Tip: For help getting started with your own data, you can schedule a free guided trial.

If you already have a Confluent Cloud Kafka cluster, you can skip step 1 and directly move on to Create an API Key. You can also skip step 3 if you already have a Confluent Cloud Kafka cluster up and running, and have created a topic that you want to create a source for.

The process to connect Materialize to a Confluent Cloud Kafka cluster consists of the following steps:

1. Create a Confluent Cloud Kafka cluster

If you already have a Confluent Cloud Kafka cluster set up, then you can skip this step.

Sign in to Confluent Cloud
Choose Create a new cluster
Select the cluster type, and specify the rest of the settings based on your needs
Choose Create cluster

Note: This creation can take about 10 minutes. For more information on the cluster creation, see Confluent Cloud documentation.

2. Create an API Key

Navigate to the Confluent Cloud dashboard
Choose the Confluent Cloud Kafka cluster you just created in Step 1
Click on the API Keys tab
In the API Keys section, choose Add Key
Specify the scope for the API key and then click Create Key. If you choose to create a granular access API key, make sure to create a service account and add an ACL with Read access to the topic you want to create a source for. Take note of the API Key you just created, as well as the API Key secret key; you’ll need them later on. Keep in mind that the API Key secret key contains sensitive information, and you should store it somewhere safe!

3. Create a topic

To start using Materialize with Confluent Cloud, you need to point it to an existing Kafka topic you want to read data from.

If you already have a topic created, you can skip this step.

Otherwise, you can find more information about how to do that here.

4. Create a connection in Materialize

Open the Confluent Cloud dashboard and select your cluster.
Click on Overview and select Cluster settings.
Copy the URL under Bootstrap server. This will be your <broker-url> going forward.
Connect to Materialize using the SQL Shell, or your preferred SQL client.
Create the connection. The exact steps depend on your networking configuration, so start by selecting the relevant option.

CREATE SECRET confluent_username AS '<your-api-key>';
CREATE SECRET confluent_password AS '<your-api-secret>';

CREATE CONNECTION confluent_cloud TO KAFKA (
    BROKER '<confluent-broker-url>',
    SASL MECHANISMS = 'PLAIN',
    SASL USERNAME = SECRET confluent_username,
    SASL PASSWORD = SECRET confluent_password
);

AWS PrivateLink lets you connect Materialize to your Confluent Cloud cluster without exposing traffic tothe public internet.

In the Confluent Cloud console, navigateto your cluster’s Networking settings and set up a PrivateLink endpoint. Record the VPC Endpoint Service Name and the DNS domain.

In the Materialize SQL shell, create a PrivateLink connectionusing the service name from the previous step. Be sure to specify all availability zones of your Confluent Cloud cluster.

CREATE CONNECTION confluent_privatelink TO AWS PRIVATELINK (
    SERVICE NAME 'com.amazonaws.vpceus-east-1vpce-svc-0e123abc123198abc',
    AVAILABILITY ZONES ('use1-az1', 'use1-az4', 'use1-az6')
);

Retrieve the AWS principal for the PrivateLink connection:

SELECT principal
FROM mz_aws_privatelink_connections plc
JOIN mz_connections c ON plc.id = c.id
WHERE c.name = 'confluent_privatelink';

In the Confluent Cloud console, add the AWS principal to thePrivateLink access list.
In Materialize, validate the PrivateLink connection:
```
VALIDATE CONNECTION confluent_privatelink;
```
If no validation error is returned, move to the next step.

Create the Kafka connection. The static broker (used forbootstrapping) does not need an AVAILABILITY ZONE — Materialize will find it across availability zones. The MATCHING rules should specify AVAILABILITYZONE to route discovered brokers through their specific AZ endpoint. The availability zones in the MATCHING rules must match the AZs whereConfluent has deployed brokers. For best results, deploy brokers across 3 AZs and select those same AZs during the Confluent PrivateLink ingress setup.

CREATE SECRET confluent_username AS '<your-api-key>';
CREATE SECRET confluent_password AS '<your-api-secret>';

CREATE CONNECTION confluent_cloud TO KAFKA (
    BROKERS (
        '<confluent-broker-url>' USING AWS PRIVATELINK      confluent_privatelink,
        MATCHING '*.use1-az1.*' USING AWSPRIVATELINKconfluent_privatelink        (AVAILABILITY ZONE='use1-az1'),
        MATCHING '*.use1-az4.*' USING AWSPRIVATELINKconfluent_privatelink        (AVAILABILITY ZONE='use1-az4'),
        MATCHING '*.use1-az6.*' USING AWSPRIVATELINKconfluent_privatelink        (AVAILABILITY ZONE ='use1-az6')
    ),
    SASL MECHANISMS = 'PLAIN',
    SASL USERNAME = SECRET confluent_username,
    SASL PASSWORD = SECRET confluent_password
);

The MATCHING patterns correspond to the AZ-specific DNS subdomains from your Confluent Cloud networking settings. Adjust the patternsand availability zones to match your cluster’s configuration.

5. Start ingesting data

Once you have created the connection, create a source and start ingesting data from your topic. By default, the source will be created in the active cluster; to use a different cluster, use the IN CLUSTER clause.

With the legacy syntax, the source decodes the topic directly and is itself queryable. Picking up an upstream schema change requires dropping and recreating the source, which incurs downtime.

CREATE SOURCE confluent_source
    FROM KAFKA CONNECTION confluent_cloud (TOPIC '')
    FORMAT JSON;

With the new syntax, create a source for the topic and then a table that decodes it. Each table pins its own reader schema, which lets you pick up upstream schema changes without downtime. For details, see Handle upstream schema changes with zero downtime.

CREATE SOURCE confluent_source
    FROM KAFKA CONNECTION confluent_cloud (TOPIC '');

CREATE TABLE confluent_table
    FROM SOURCE confluent_source
    FORMAT JSON;

If the command executes without an error, it means that you have successfully connected Materialize to your Confluent Cloud Kafka cluster.

Note: The examples above use JSON, but you can also ingest Kafka messages formatted in other supported formats; e.g., Avro and CSV. You can find more details about the various different supported formats and possible configurations in the reference documentation.