Kafka

Kafka Reader 2.1 (using Apache Kafka client 2.1) is deprecated as of Striim 5.4.0.2. When upgrading to 5.4.0.2, any Kafka Reader 2.1 adapters in created or quiesced applications are automatically upgraded to Kafka Reader 3.9 and corresponding connection profiles are created.

If you import TQL exported from Striim 5.2 or earlier after upgrading, any Kafka Reader components will use the deprecated Kafka Reader 2.1. To use the current version, edit the TQL before importing to use the new properties, and create a connection profile with the appropriate settings.

The following table shows how properties map from Kafka Reader 2.1 to the current Kafka Reader.

Kafka Reader requires a Kafka connection profile for broker address, authentication, and security configuration. Create a connection profile before configuring the reader.

A single Kafka connection profile can be shared between Kafka Reader and Kafka Writer components. Each component uses a different subset of the connection profile properties.

Kafka Reader can consume only records that are part of committed Kafka transactions by setting the consumer isolation level in the connection profile.

To read only committed messages, set the following in the Consumer Config section of the connection profile:

isolation.level=read_committed

The default value (if not set) is read_uncommitted. When using read_committed, records from aborted transactions are filtered out by the consumer. Offsets in the log may appear to skip values because Kafka assigns offsets to all records, including aborted ones that are not returned to the reader.

When starting from a specific offset, consumption begins at that offset or the next visible committed record, whichever applies. If the specified offset falls inside an aborted transaction, the reader advances to the next committed record.

Kafka Reader supports Avro record keys with schema registry integration, enabling richer key structures beyond traditional string keys. Features include multi-field keys, nested key objects, and automatic key schema resolution from Confluent or Karapace Schema Registry.

To configure Avro key support, add the key.deserializer property to the Consumer Config section of the connection profile:

key.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer

Also configure the schema registry URL in the Avro Parser properties.

Karapace is an open-source schema registry that is API-compatible with Confluent Schema Registry. Kafka Reader supports Karapace via direct URL configuration in the Avro Parser.

Configure the following Avro Parser properties for Karapace:

Example Avro Parser configuration with Karapace and BASIC authentication:

PARSE USING Global.AvroParser (
  schemaregistryurl: 'https://karapace.example.com:8081',
  schemaregistryConfiguration:
    'basic.auth.user.info=username:password,basic.auth.credentials.source=USER_INFO'
)

Kafka Reader can consume Data Definition Language (DDL) changes that originate from OLTP sources and are produced by Striim targets. Kafka Reader treats DDL as normal Kafka records, exposing headers, key, timestamp, and offsets in WAEvent.metadata and passing the payload to the parser.

To identify DDL records in the parser output, check the OperationType field for a value of DDL (with OperationName values such as CREATE, ALTER, or DROP).

Example filter query:

SELECT *
FROM KafkaInput
WHERE ddl_operation IN ("CREATE","ALTER","DROP");

If you only want committed DDL/DML, set isolation.level=read_committed via the connection profile Consumer Config.

Kafka Reader supports seamless certificate rotation via connection profiles. New server and client certificate pairs can be generated and applied to a Kafka connection profile even when applications using the profile are running.

The Connection Retry Policy automatically handles temporary connection failures during certificate rotation. When certificates are updated, the reader reconnects under the configured retry policy. Jitter is applied to prevent multiple Kafka Readers from retrying simultaneously during rotation events.

To rotate certificates:

Kafka Reader exposes the following metadata fields on each WAEvent:

When Kafka Writer has been configured to add transaction context to record headers, the KafkaRecordHeader map may contain TxnID, TableName, and OperationName fields.

During retry operations, Kafka Reader performs offset validation that may result in fatal exceptions that cannot be retried or ignored.

Kafka Reader supports the following parsers.

The following sample application reads data from a Kafka topic using a connection profile and writes it to a file:

CREATE SOURCE KafkaReaderSample USING KafkaReader (
  connectionProfileName: 'admin.KafkaCP',
  topic: 'KafkaWriterSample',
  startOffset: '0'
)
PARSE USING DSVParser ()
OUTPUT TO RawKafkaStream;

CREATE TARGET KafkaReaderOut USING FileWriter (
  filename: 'KafkaReaderOutput'
)
FORMAT USING DSVFormatter ()
INPUT FROM RawKafkaStream;

Striim uses property sets to configure Kafka connections for persistent streams and other Kafka-related functionality. In Striim 5.4.0, the Kafka property set syntax has been updated to remove the dependency on Zookeeper and align with the official Apache Kafka API.

CREATE PROPERTYSET KafkaPS ('bootstrap.servers': 'localhost:9092', partitions: '250');

CREATE PROPERTYSET KafkaPS (bootstrap.brokers: 'localhost:9092', partitions: '250');

-- This syntax continues to work but zk.address is no longer required
CREATE PROPERTYSET KafkaPS (zk.address: 'localhost:2181', bootstrap.brokers: 'localhost:9092', partitions: '250');

Kafka Reader uses a Kafka connection profile for all authentication and security configuration. The connection profile supports NONE, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI (Kerberos), Mutual TLS, and AWS MSK IAM authentication methods.

For an overview of this feature, see Introducing Kafka streams.

Reading from a Kafka stream can be useful for development tasks such as doing A/B comparisons of variations on a TQL application. If you modified Samples.PosApp.tql to persist PosDataStream to Kafka, the following source would read the persisted data from Kafka.

CREATE SOURCE AccessLogSource USING KafkaReader (
  connectionProfileName: 'admin.KafkaCP',
  Topic: 'Samples_PosDataStream',
  PartitionIDList: '0',
  startOffset: 0
)
PARSE USING StriimParser ()
OUTPUT TO KafkaDSVStream;

For more information, see Kafka.Kafka Reader