Kafka and Redpanda

Connectors

Materialize bundles a native connector that allow writing data to Kafka and Redpanda.

For details on the connector, including syntax, supported formats and examples, refer to CREATE SINK.

Features

Automatic topic creation

If the specified Kafka topic does not exist, Materialize will attempt to create it using the broker’s default number of partitions, default replication factor, default compaction policy, and default retention policy, unless any specific overrides are provided as part of the connection options.

If the connection’s progress topic does not exist, Materialize will attempt to create it with a single partition, the broker’s default replication factor, compaction enabled, and both size- and time-based retention disabled. The replication factor can be overridden using the PROGRESS TOPIC REPLICATION FACTOR option when creating a connection CREATE CONNECTION.

To customize topic-level configuration, including compaction settings and other values, use the TOPIC CONFIG option in the connection options to set any relevant kafka topic configs.

If you manually create the topic or progress topic in Kafka before running CREATE SINK, observe the following guidance:

Exactly-once processing

By default, Kafka sinks provide exactly-once processing guarantees, which ensures that messages are not duplicated or dropped in failure scenarios.

To achieve this, Materialize stores some internal metadata in an additional progress topic. This topic is shared among all sinks that use a particular Kafka connection. The name of the progress topic can be specified when creating a connection; otherwise, a default name of _materialize-progress-{REGION ID}-{CONNECTION ID} is used. In either case, Materialize will attempt to create the topic if it does not exist. The contents of this topic are not user-specified.

End-to-end exactly-once processing

Exactly-once semantics are an end-to-end property of a system, but Materialize only controls the initial produce step. To ensure end-to-end exactly-once message delivery, you should ensure that:

• The broker is configured with replication factor greater than 3, with unclean leader election disabled (unclean.leader.election.enable=false).
• All downstream consumers are configured to only read committed data (isolation.level=read_committed).
• The consumers’ processing is idempotent, and offsets are only committed when processing is complete.

For more details, see the Kafka documentation.

Partitioning

By default, Materialize assigns a partition to each message using the following strategy:

1. Encode the message’s key in the specified format.
2. If the format uses a Confluent Schema Registry, strip out the schema ID from the encoded bytes.
3. Hash the remaining encoded bytes using [SeaHash].
4. Divide the hash value by the topic’s partition count and assign the remainder as the message’s partition.

If a message has no key, all messages are sent to partition 0.

To configure a custom partitioning strategy, you can use the PARTITION BY option. This option allows you to specify a SQL expression that computes a hash for each message, which determines what partition to assign to the message:

-- General syntax.
CREATE SINK ... INTO KAFKA CONNECTION <name> (PARTITION BY = <expression>) ...;

-- Example.
CREATE SINK ... INTO KAFKA CONNECTION <name> (
    PARTITION BY = kafka_murmur2(name || address)
) ...;

The expression:

• Must have a type that can be assignment cast to [uint8].
• Can refer to any column in the sink’s underlying relation when using the upsert envelope.
• Can refer to any column in the sink’s key when using the Debezium envelope.

Materialize uses the computed hash value to assign a partition to each message as follows:

1. If the hash is NULL or computing the hash produces an error, assign partition 0.
2. Otherwise, divide the hash value by the topic’s partition count and assign the remainder as the message’s partition (i.e., partition_id = hash % partition_count).

Materialize provides several hash functions which are commonly used in Kafka partition assignment:

• crc32
• kafka_murmur2
• seahash

For a full example of using the PARTITION BY option, see Custom partioning.

Kafka transaction markers