CREATE SOURCE: Kafka/Redpanda (New Syntax)

Creates a new source from Kafka or Redpanda broker. Once a new source is created, you can CREATE TABLE FROM SOURCE to create the corresponding tables in Materialize and start the data ingestion process.

The decoding options (FORMAT, INCLUDE, and ENVELOPE) are set on the CREATE TABLE ... FROM SOURCE statement that reads from the source. For the full catalog of formats, envelopes, and exposed metadata, see CREATE TABLE: Kafka source table.

Prerequisites

To create a source from Kafka/Redpanda broker, you first need to create a connection. Once created, a connection is reusable across multiple CREATE SOURCE and CREATE SINK statements.

Syntax

The CREATE SOURCE statement connects to a Kafka/Redpanda topic.

CREATE SOURCE [IF NOT EXISTS] <src_name>
[IN CLUSTER <cluster_name>]
FROM KAFKA CONNECTION <connection_name> (
  TOPIC '<topic>'
  [, GROUP ID PREFIX '<group_id_prefix>']
  [, START OFFSET ( <partition_offset> [, ...] ) ]
  [, START TIMESTAMP <timestamp> ]
)
[EXPOSE PROGRESS AS <progress_subsource_name>]
[WITH ( <with_option> [, ...] )];

Details

Ingesting data

After the source is created, each CREATE TABLE ... FROM SOURCE statement creates a table that decodes the topic and starts ingesting data. You can create multiple tables from the same source, each with its own format and envelope.

Handling schema changes

Because each table pins its own reader schema when it is created, you can pick up a compatible upstream schema change without downtime: create a new table that reads the evolved schema, recreate the downstream objects, and swap them into place. See Handle upstream schema changes with zero downtime for the full procedure.

Features

Setting start offsets

To start consuming a Kafka stream from a specific offset, you can use the START OFFSET option.

CREATE SOURCE kafka_offset
  FROM KAFKA CONNECTION kafka_connection (
    TOPIC 'data',
    -- Start reading from the earliest offset in the first partition,
    -- the second partition at 10, and the third partition at 100.
    START OFFSET (0, 10, 100)
  );

Note that:

• If fewer offsets than partitions are provided, the remaining partitions will start at offset 0. This is true if you provide START OFFSET (1) or START OFFSET (1, ...).

• Providing more offsets than partitions is not supported.

Time-based offsets

It’s also possible to set a start offset based on Kafka timestamps, using the START TIMESTAMP option. This approach sets the start offset for each available partition based on the Kafka timestamp and the source behaves as if START OFFSET was provided directly.

It’s important to note that START TIMESTAMP is a property of the source: it will be calculated once at the time the CREATE SOURCE statement is issued. This means that the computed start offsets will be the same for all views depending on the source and stable across restarts.

If you need to limit the amount of data maintained as state after source creation, consider using temporal filters instead.

Monitoring source progress

By default, Kafka sources expose progress metadata as a subsource that you can use to monitor source ingestion progress. The name of the progress subsource can be specified when creating a source using the EXPOSE PROGRESS AS clause; otherwise, it will be named <src_name>_progress.

The following metadata is available for each source as a progress subsource:

And can be queried using:

SELECT
  partition, "offset"
FROM
  (
    SELECT
      -- Take the upper of the range, which is null for non-partition rows
      -- Cast partition to u64, which is more ergonomic
      upper(partition)::uint8 AS partition, "offset"
    FROM
      <src_name>_progress
  )
WHERE
  -- Remove all non-partition rows
  partition IS NOT NULL;

As long as any offset continues increasing, Materialize is consuming data from the upstream Kafka broker. For more details on monitoring source ingestion progress and debugging related issues, see Troubleshooting.

Monitoring consumer lag

To support Kafka tools that monitor consumer lag, Kafka sources commit offsets once the messages up through that offset have been durably recorded in Materialize’s storage layer.

However, rather than relying on committed offsets, Materialize suggests using our native progress monitoring, which contains more up-to-date information.

Committed offsets are associated with a consumer group specific to the source. The ID of the consumer group consists of the prefix configured with the GROUP ID PREFIX option followed by a Materialize-generated suffix.

You should not make assumptions about the number of consumer groups that Materialize will use to consume from a given source. The only guarantee is that the ID of each consumer group will begin with the configured prefix.

The consumer group ID prefix for each Kafka source in the system is available in the group_id_prefix column of the [mz_kafka_sources] table. To look up the group_id_prefix for a source by name, use:

SELECT group_id_prefix
FROM mz_internal.mz_kafka_sources ks
JOIN mz_sources s ON s.id = ks.id
WHERE s.name = '<src_name>'

For spilling to disk, see the Features section of the Kafka/Redpanda reference page. This feature is configured on the CREATE SOURCE statement and behaves the same regardless of syntax.

Examples

Prerequisite: Creating a connection

A connection describes how to connect and authenticate to an external system you want Materialize to read data from.

Once created, a connection is reusable across multiple CREATE SOURCE statements. For more details on creating connections, check the CREATE CONNECTION documentation page.

Broker

CREATE SECRET kafka_ssl_key AS '<BROKER_SSL_KEY>';
CREATE SECRET kafka_ssl_crt AS '<BROKER_SSL_CRT>';

CREATE CONNECTION kafka_connection TO KAFKA (
    BROKER 'unique-jellyfish-0000.us-east-1.aws.confluent.cloud:9093',
    SSL KEY = SECRET kafka_ssl_key,
    SSL CERTIFICATE = SECRET kafka_ssl_crt
);

CREATE SECRET kafka_password AS '<BROKER_PASSWORD>';

CREATE CONNECTION kafka_connection TO KAFKA (
    BROKER 'unique-jellyfish-0000.us-east-1.aws.confluent.cloud:9092',
    SASL MECHANISMS = 'SCRAM-SHA-256',
    SASL USERNAME = 'foo',
    SASL PASSWORD = SECRET kafka_password
);

If your Kafka broker is not exposed to the public internet, you can tunnel the connection through an AWS PrivateLink service (Materialize Cloud) or an SSH bastion host:

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

CREATE CONNECTION kafka_connection TO KAFKA (
    BROKERS (
        'broker1:9092' USING AWS PRIVATELINK privatelink_svc,
        'broker2:9092' USING AWS PRIVATELINK privatelink_svc (PORT 9093)
    )
);

CREATE CONNECTION ssh_connection TO SSH TUNNEL (
    HOST '<SSH_BASTION_HOST>',
    USER '<SSH_BASTION_USER>',
    PORT <SSH_BASTION_PORT>
);

CREATE CONNECTION kafka_connection TO KAFKA (
BROKERS (
    'broker1:9092' USING SSH TUNNEL ssh_connection,
    'broker2:9092' USING SSH TUNNEL ssh_connection
    )
);

Confluent Schema Registry

CREATE SECRET csr_ssl_crt AS '<CSR_SSL_CRT>';
CREATE SECRET csr_ssl_key AS '<CSR_SSL_KEY>';
CREATE SECRET csr_password AS '<CSR_PASSWORD>';

CREATE CONNECTION csr_connection TO CONFLUENT SCHEMA REGISTRY (
    URL 'https://unique-jellyfish-0000.us-east-1.aws.confluent.cloud:9093',
    SSL KEY = SECRET csr_ssl_key,
    SSL CERTIFICATE = SECRET csr_ssl_crt,
    USERNAME = 'foo',
    PASSWORD = SECRET csr_password
);

CREATE SECRET IF NOT EXISTS csr_username AS '<CSR_USERNAME>';
CREATE SECRET IF NOT EXISTS csr_password AS '<CSR_PASSWORD>';

CREATE CONNECTION csr_connection TO CONFLUENT SCHEMA REGISTRY (
  URL '<CONFLUENT_REGISTRY_URL>',
  USERNAME = SECRET csr_username,
  PASSWORD = SECRET csr_password
);

If your Confluent Schema Registry server is not exposed to the public internet, you can tunnel the connection through an AWS PrivateLink service (Materialize Cloud) or an SSH bastion host:

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

CREATE CONNECTION csr_connection TO CONFLUENT SCHEMA REGISTRY (
    URL 'http://my-confluent-schema-registry:8081',
    AWS PRIVATELINK privatelink_svc
);

CREATE CONNECTION ssh_connection TO SSH TUNNEL (
    HOST '<SSH_BASTION_HOST>',
    USER '<SSH_BASTION_USER>',
    PORT <SSH_BASTION_PORT>
);

CREATE CONNECTION csr_connection TO CONFLUENT SCHEMA REGISTRY (
    URL 'http://my-confluent-schema-registry:8081',
    SSH TUNNEL ssh_connection
);

AWS Glue Schema Registry

An AWS Glue Schema Registry connection authenticates through a separate AWS connection, which supplies the credentials and region:

CREATE CONNECTION aws_connection TO AWS (
    ASSUME ROLE ARN = 'arn:aws:iam::123456789000:role/MaterializeGlue'
);

CREATE CONNECTION glue_connection TO AWS GLUE SCHEMA REGISTRY (
    AWS CONNECTION = aws_connection,
    REGISTRY = 'default-registry'
);

The AWS connection must be allowed to read schemas from the registry. See Permissions for the required IAM actions.

Create a source and table

CREATE SOURCE orders_src
  FROM KAFKA CONNECTION kafka_connection (TOPIC 'orders');

CREATE TABLE orders
  FROM SOURCE orders_src
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection
  ENVELOPE UPSERT;

For connection setup, required Kafka ACLs, and worked examples for each format, see the Kafka/Redpanda reference page.

Related pages

• CREATE TABLE
• CREATE SECRET
• CREATE CONNECTION
• CREATE SOURCE: Kafka/Redpanda (Legacy Syntax)
• Handle upstream schema changes with zero downtime