Apache Kafka with DLQ and metadata¶

Key features¶

The Apache Kafka with DLQ and metadata connector provides comprehensive functionality:

• Dead Letter Queue (DLQ) support for failed message handling

• RECORD_METADATA column with Kafka message metadata

• Configurable schematization - enable or disable schema detection

• Iceberg table support with schema evolution

• Multiple message formats - JSON and AVRO support

• Schema registry integration for AVRO messages

• Topic-to-table mapping with advanced patterns

• SASL authentication support

Specific parameters¶

In addition to the common parameters described in Set up the Openflow Connector for Kafka, this connector includes additional parameter contexts for advanced features.

Schematization behavior¶

The connector’s behavior changes based on the Schematization Enabled parameter:

Schema detection and evolution¶

The connector supports schema detection and evolution. The structure of tables in Snowflake can be defined and evolved automatically to support the structure of new data loaded by the connector.

Without schema detection and evolution, the Snowflake table loaded by the connector only consists of two OBJECT columns: RECORD_CONTENT and RECORD_METADATA.

With schema detection and evolution enabled, Snowflake can detect the schema of the streaming data and load data into tables that automatically match any user-defined schema. Snowflake also allows adding new columns or dropping the NOT NULL constraint from columns missing in new data files.

Schema detection with the connector is supported with or without a provided schema registry. If using schema registry (Avro), the column will be created with the data types defined in the provided schema registry. If there is no schema registry (JSON), the data type will be inferred based on the data provided.

JSON ARRAY is not supported for further schematization.

Dead Letter Queue (DLQ)¶

The DLQ functionality handles messages that cannot be processed successfully:

Iceberg table support¶

Openflow Connector for Kafka can ingest data into a Snowflake-managed Apache Iceberg™ table when Iceberg Enabled is set to true.

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    record_metadata OBJECT()
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';

{
    "id": 1,
    "name": "Steve",
    "body_temperature": 36.6,
    "approved_coffee_types": ["Espresso", "Doppio", "Ristretto", "Lungo"],
    "animals_possessed":
    {
        "dogs": true,
        "cats": false
    },
    "date_added": "2024-10-15"
}

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    RECORD_METADATA OBJECT(
        offset INTEGER,
        topic STRING,
        partition INTEGER,
        key STRING,
        timestamp TIMESTAMP,
        SnowflakeConnectorPushTime BIGINT,
        headers MAP(VARCHAR, VARCHAR)
    ),
    id INT,
    body_temperature FLOAT,
    name STRING,
    approved_coffee_types ARRAY(STRING),
    animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
    date_added DATE
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    RECORD_METADATA OBJECT(
        offset INTEGER,
        topic STRING,
        partition INTEGER,
        key STRING,
        timestamp TIMESTAMP,
        SnowflakeConnectorPushTime BIGINT,
        headers MAP(VARCHAR, VARCHAR)
    ),
    RECORD_CONTENT OBJECT(
        id INT,
        body_temperature FLOAT,
        name STRING,
        approved_coffee_types ARRAY(STRING),
        animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
        date_added DATE
    )
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';

Use cases¶

This connector is ideal for:

• Production environments requiring DLQ

• Data lineage and auditing where Kafka metadata is important

• Complex message processing with schema evolution requirements

• Iceberg table integration

If you need simpler ingestion without metadata or DLQ features, consider the Apache Kafka for JSON/AVRO data format connectors instead.