Data Formatting
Avro Formatter
When using the Avro Formatter, an Avro record is generated corresponding to the incoming event, and respective schemas are registered in the schema registry. See Data Type Mapping for how source column types map to Avro types. Three formatting modes are available via the formatAs property:
Default: An Avro record is generated with metadata, userdata, data, and before fields of WAEvent converted into Maps of type String. Preferred for all events from non-OLTP/DWH sources, or for OLTP/DWH when preserving source data types is not required.
Native: metadata, userdata, data, before, data bit presence, and before bit presence fields are all considered. Data and before fields are nested Avro records with the exact field count. Names, types, and aliases are mapped from the incoming event's Striim TYPE. Userdata/metadata are Map(String) fields.
Table: Only the data field of WAEvent is considered, with the exact fields for the respective schema. Name and type are mapped from the incoming event's Striim TYPE.
Avro Record Serialization
By default, Kafka Writer uses the Striim Serializer for Avro-formatted messages. Set Serializer: ConfluentSerializer on Kafka Writer to use the Confluent Serializer instead, when downstream consumers are using standard Kafka/Confluent-ecosystem tooling.
Striim Serializer (default)
Each Kafka message value is framed as:
[4-byte payload length][payload], where payload = [4-byte schema registry ID — only if a schema registry is configured][Avro-encoded record bytes]
This is Striim's own framing, not Confluent's. To read it back from a plain Java consumer (rather than a Striim Kafka Reader), set value.deserializer to the matching class for how the data was written:
Written with | value.deserializer |
|---|---|
Schema registry configured | com.striim.kafka.deserializer.KafkaAvroDeserializer |
Schema file only (no registry) | com.striim.kafka.deserializer.StriimAvroLengthDelimitedDeserializer |
Reading Striim-serialized Avro back into Striim itself (Kafka Reader + Avro Parser) needs no extra configuration — com.striim.avro.deserializer.LengthDelimitedAvroRecordDeserializer is already the Kafka Reader default for both variants; only the Avro Parser's SchemaRegistryURL or SchemaFileName needs to match how it was written.
Confluent Serializer
Set Serializer: ConfluentSerializer to use Confluent's standard wire format: a single 0x0 magic byte, a 4-byte schema ID, then the Avro-encoded record bytes — no separate length prefix. Striim delegates to Confluent's own io.confluent.kafka.serializers.KafkaAvroSerializer to produce this, so there's no Striim-specific deviation from Confluent's documented format. A schema registry connection is required.
To read it back from a plain Java consumer, set value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer — Confluent's own deserializer, not a Striim class. The same applies when reading Confluent-format data back into Striim via Kafka Reader + Avro Parser.
Choosing between them
Striim Serializer (default) | Confluent Serializer | |
|---|---|---|
Wire format | 4-byte length prefix + optional 4-byte schema ID + Avro bytes | Confluent standard: magic byte + 4-byte schema ID + Avro bytes |
Schema registry | Optional | Required |
Readable by generic Confluent-ecosystem consumers | No — requires Striim's deserializer classes | Yes — standard KafkaAvroDeserializer |
Best for | Striim-to-Striim pipelines | Confluent Schema Registry–integrated deserializers, Kafka Connect, ksqlDB, third-party Avro clients |
Avro Formatter supports:
Confluent Serializer (set via the Serializer property) in addition to the default Striim Serializer.
Schema compatibility configuration via the SchemaRegistryCompatibility property (default: None).
Subject naming strategies: default, UseTopicName or UseDynamicValues.
The table below lists the AvroFormatter properties that are new or changed for Kafka Writer. Existing AvroFormatter properties not listed here are unchanged.
Property | Required / Default | Description |
|---|---|---|
Use Schema Registry Connection Profile | Required: False. Default: False. | Toggle to enable the Schema Registry Connection Profile. When true, the schema registry URL and schema registry configuration properties are not accepted. |
Schema Registry Connection Profile Name | Required: False (True when Use Schema Registry Connection Profile = true). | Name of the Schema Registry Connection Profile to use. Currently only Confluent Schema Registry is supported via Connection Profile. |
Avro Record Namespace | Required: False. Default: uses the Avro record name. | Logically groups related record types within an Avro schema. Value can be @metadata(<field>), @userdata(<field>), a static quoted name, or an incoming field name for typed events. |
Schema Registry SubjectName | Required: False. Default: namespace + Avro record name. | Name of the subject the Avro schema is registered under. Accepts UseTopicName (all records on a topic share one subject) or UseDynamicValues (each record type gets its own subject, from the field named in Schema Registry SubjectName Mapping). |
Schema Registry SubjectName Mapping | Required: True when SubjectName = UseDynamicValues. | Field used to derive the subject name: @metadata(<field>), @userdata(<field>), a static quoted name, or an incoming field name for typed events. |
Schema Compatibility | Required: False. Default: None. | Compatibility mode is enforced when the schema evolves. See Schema Compatibility below for all supported modes. Subject Name Mapping determines the subject under which each Avro schema is registered, and has three scenarios: if left empty, the subject is <namespace>.<record name>, with the message key registered under <source table>-key, the value under <source table>-value, and (if PersistSchema is on) the DDL schema under <source table>-DDLRecord / <source table>-DDLKey; with UseTopicName, all Avro records published to a topic share one subject, using <topic>-key / <topic>-value / <topic>-DDLRecord / <topic>-DDLKey; with UseDynamicValues, the subject is picked from the field named in Schema Registry SubjectName Mapping, using the same -key / -value / -DDLRecord / -DDLKey suffix pattern. |
Avro Record Structure
An Avro record consists of the following key parts:
Name: uniquely identifies the record type within its namespace. Example: "name": "Employee"
Namespace: logically groups related record types within an Avro schema. Example: "namespace": "com.company.hr"
Type: specifies the schema type (String, Avro Record type, Union, Map, etc.)
Fields: a list of field definitions within the record. Each field has:
Name: identifier for the field (e.g., "firstName")
Type: data type or union of types (e.g., "string", ["null", "int"])
Default (optional): default value if not provided in the data
Doc (optional): description of the field, for schema documentation
Aliases (optional): alternate names for the record, to support schema evolution.
With default namespace and subject-name-mapping settings, the resulting Avro record — across the various formatAs settings — closely matches existing Kafka Writer/Avro Formatter output. The one change: the __striimmetadata field is no longer present in any Avro record format.
Once the Avro record is constructed, its schema is registered in the configured schema registry, under the subject name set by the Schema Registry SubjectName property (see Subject Name Mapping below).
Avro Record Name
Default values for the Avro record name, by event type:
CDC sources (WAEvent):
formatAs: Native or Table — applies only to OLTP sources, which use two-part (schema.table) or three-part (catalog.schema.table) names. The record name is always the table name (the last segment of the fully qualified name), taken by default from the event metadata field TableName. If Schema Registry SubjectName is left at its default, the record's namespace + name together form the subject name under which the schema is registered; if set to Topic or Custom, the record name and subject name will differ.
formatAs: Default — uses "WAEvent" as the record name. Applies to non-OLTP sources sending WAEvent.
AvroEvent:
If the incoming event has only an Avro record in the data field, uses the source Avro record's namespace.name as the record name.
If the incoming event has Data and Metadata, uses the topic name as the record name.
JsonNodeEvent: uses the topic name as the record name.
ParquetEvent: uses "ParquetEvent" as the record name.
Typed Event: uses the typed stream name as the record name.
Note
only underscore and hyphen are supported as special characters in the Avro record name — any other special character can cause the application to HALT.
Avro Record Namespace Mapping
Namespace in Avro groups related schemas and prevents naming conflicts. The Avro Formatter supports namespace mapping across all three formatAs types (Native, Table, Default) and for all incoming event types.
Default behavior:
Uses the value from Avro Record Name and appends "_namespace".
If the record name contains periods (e.g., Catalog.Sch.EMP), everything before the last period becomes the namespace, and the final segment becomes the record name.
FormatAs / Source | Scenario | Resulting namespace + name |
|---|---|---|
Default | WAEvent | "namespace": "WAEvent_namespace", "name": "WAEvent" |
Default | AvroEvent | "namespace": "EMPTopic_namespace", "name": "EMPTopic" |
Default | JSONNodeEvent | "namespace": "EMPTopic_namespace", "name": "EMPTopic" |
Default | ParquetEvent | "namespace": "ParquetEvent_namespace", "name": "ParquetEvent" |
Default | Typed Event (e.g. SalesTransaction) | "namespace": "SalesTransaction_namespace", "name": "SalesTransaction" |
Table / Native | WAEvent, OLTP source, table Sch.EMP | "namespace": "Sch", "name": "EMP" (schema name of the table is used) |
Table / Native | WAEvent, OLTP source, table Catalog.Sch.EMP | "namespace": "Catalog.Sch", "name": "EMP" (database + schema name is used) |
Dynamic namespace values: Each Avro record type can be registered under its own namespace if a dynamic value is specified via the Avro Record Namespace property. The namespace is picked from a field specified in the Avro Formatter and applies to all records from that Kafka Writer. Accepted sources: a field name (for typed input streams), a static value, or @metadata(<field>) / @userdata(<field>) (for WAEvent, AvroEvent, JSONNodeEvent, ParquetEvent).
Example: with AvroRecord Name = "Emp" and Custom AvroRecord Namespace = "company.hr.schema":
{ "namespace": "company.hr.schema", "type": "record", "name": "Emp", "fields": [ { … } ] }Note
if the Avro record name contains a period and AvroRecordNamespace is set to a custom value, only the last segment after the final period is used as the record name — any segments before it are ignored — and the custom value is used as the namespace as-is.
Namespace mapping determines the subject name under which schemas are registered if Subject Name Mapping is left empty (see Subject Name Mapping below).
Note
supported special characters in the Avro record namespace are underscore (_), period (.), and hyphen (-). Any other special character in the resulting record name or namespace can cause the application to HALT.
Subject Name Mapping
Avro subject name mapping determines the subject name under which each Avro record's schema is registered in the schema registry. This affects schema reuse, evolution, and compatibility across topics. Supported for all three formatAs types (Native, Table, Default) and for all incoming event types.
Scenario 1: Left empty (default)
Subject name = <avro record namespace>.<avro record name> (name/namespace derived per event type as described above; the namespace can be overridden via the Avro Record Namespace property).
For formatAs: Native or Table — regardless of whether the topic mapping is a wildcard with no prefix/suffix (e.g. %,%), a wildcard with a prefix/suffix (e.g. %,%_uwm), or a direct mapping (e.g. sourcetable,tgttable) — the subject is the source table name. Message key subject = <source table>-key; message value subject = <source table>-value. If PersistSchema is ON, the DDL message value schema registers under <source table>-DDLRecord, and the DDL message key schema under <source table>-DDLKey.
For formatAs: Default, see the per-event-type breakdown table below.
Scenario 2: UseTopicName
All Avro records map to the Kafka topic name as their subject. If multiple Avro record types publish to the same topic, that one subject accumulates multiple schema versions.
Message key subject = <topic name>-key; message value subject = <topic name>-value. If PersistSchema is ON, DDL message value schema = <topic name>-DDLRecord, DDL message key schema = <topic name>-DDLKey.
Scenario 3: UseDynamicValues
Each Avro record type registers under its own subject, dynamically picked from a field specified via the Schema Registry SubjectName Mapping property.
Message key subject = <field-specific value>-key; message value subject = <field-specific value>-value. If PersistSchema is ON, DDL message value schema = <field-specific value>-DDLRecord, DDL message key schema = <field-specific value>-DDLKey.
Note
for AvroEvent, JsonNodeEvent, and Typed Event, Default is the only supported formatAs option.
Detailed breakdown by FormatAs and event type
Scenario 1 — Subject name mapping left empty
formatAs | Event type | Message value subject | Message key subject | Avro record name | Avro record namespace |
|---|---|---|---|---|---|
Default | WAEvent | WAEvent | WAEvent-Key | WAEvent | WAEvent_namespace (or custom) |
Default | Typed Event | Incoming stream name | <stream name>-Key | Incoming stream name | <stream name>_namespace (or custom) |
Default | JSONNodeEvent | Topic name | <topic name>-Key | Topic name | <topic name>_namespace (or custom) |
Default | AvroEvent | Topic name | <topic name>-Key | Topic name | <topic name>_namespace (or custom) |
Default | ParquetEvent | ParquetEvent | ParquetEvent-Key | ParquetEvent | ParquetEvent_namespace (or custom) |
Native (OLTP only) | WAEvent | <fully qualified table>-value (e.g. sch.emp-value) | <table>-key (e.g. sch.emp-key) | Table name | [catalog.]schema of the table (or custom) |
Table (OLTP only) | WAEvent | <fully qualified table>-value (e.g. sch.emp-value) | <table>-key (e.g. sch.emp-key) | Table name | [catalog.]schema of the table (or custom) |
Native and Table are supported only when the source is OLTP and the incoming event is WAEvent — not supported for any other event type.
For Native/Table: when SchemaEvolution: Auto and the incoming WAEvent is a DDL operation with PersistSchema ON, the DDL message value subject = <fully qualified table>-DDLRecord and the DDL message key subject = <fully qualified table>-DDLKey.
Scenario 2 — UseTopicName
formatAs | Event type | Subject name |
|---|---|---|
Default | WAEvent | Message key: <mapped topic>-key; message value: <mapped topic>-value. Avro record name = WAEvent, namespace = WAEvent_namespace (or custom). |
Default | JSONNodeEvent | Mapped topic name |
Default | Typed Event | Mapped topic name |
Default | AvroEvent | Mapped topic name |
Native (OLTP only) | WAEvent | Message key: <mapped topic>-key; message value: <mapped topic>-value. Avro record name = table name, namespace = [catalog.]schema (or custom). |
Table (OLTP only) | WAEvent | Same as Native above. |
For JsonNodeEvent, AvroEvent, ParquetEvent, and Typed Event, the Avro record name and namespace follow the same format as Scenario 1 (default subject name mapping).
Native and Table are supported only for OLTP sources with WAEvent — not supported for any other event type.
With SchemaEvolution: Auto and PersistSchema ON for a WAEvent DDL operation: DDL message value subject = <mapped topic>-DDLRecord; DDL message key subject = <mapped topic>-DDLKey.
Scenario 3 — UseDynamicValues
formatAs | Event type | Subject name |
|---|---|---|
Default | WAEvent | Message value: <field-specific value>-value; message key: <field-specific value>-Key. Avro record name = WAEvent, namespace = WAEvent_namespace (or custom). |
Default | Typed Event | Message value: <field-specific value>-value; message key: <field-specific value>-Key. |
Default | JSONNodeEvent | Message value: <field-specific value>-value; message key: <field-specific value>-Key. |
Default | AvroEvent | Message value: <field-specific value>-value; message key: <field-specific value>-Key. |
Native (OLTP only) | WAEvent | Message value: <field-specific value>-value; message key: <field-specific value>-Key. Avro record name = table name, namespace = [catalog.]schema (or custom). |
Table (OLTP only) | WAEvent | Same as Native above. |
For JsonNodeEvent, AvroEvent, ParquetEvent, and Typed Event, the Avro record name and namespace follow the same format as Scenario 1 (default subject name mapping).
Native and Table are supported only for OLTP sources with WAEvent — not supported for any other event type.
With SchemaEvolution: Auto and PersistSchema ON for a WAEvent DDL operation: DDL message value subject = <field-specific value>-DDLRecord; DDL message key subject = <field-specific value>-DDLKey.
Note
for formatAs: Default, existing 5.2.0 behavior is preserved — changing this subject name mapping strategy would change Kafka Writer/Avro Formatter behavior for existing users.
Schema Handling
Initial Schema Creation
To achieve closer Avro data type mapping, migrate the source schema before starting the Kafka Writer. Striim provides three ways to create the initial schema:
Using Database Reader
Run an initial-load-only application that flows the source schema through the pipeline. Configure Database Reader with CreateSchema: true and QuiesceOnILCompletion: true. Configure Kafka Writer with:
TopicKey: @metadata(TableName)
AutoCreateTopic: False
SchemaEvolution: Auto
PersistDDL: On
AvroFormatter with formatAs: Native (or Table) and SchemaRegistrySubjectName: UseTopicName
CREATE OR REPLACE APPLICATION demo RECOVERY 5 SECOND INTERVAL; CREATE SOURCE s1 USING Global.DatabaseReader ( DatabaseProviderType: 'Oracle', Username: 'qatest', Password: '<password>', ConnectionURL: 'jdbc:oracle:thin:@//localhost:1521/orcl', Tables: 'QATEST.EMPLOYEES', FetchSize: 100, ParallelThreads: 1, CreateSchema: true, QuiesceOnILCompletion: true, RestartBehaviourOnILInterruption: 'keepTargetTableData' ) OUTPUT TO STRIIM1; CREATE OR REPLACE TARGET t1 USING Global.KafkaWriter ( ConnectionProfileName: 'admin.Kafka_Connection_Profile', Topics: '%,%', TopicKey: '@metadata(TableName)', AutoCreateTopic: true, MessageKey: 'PrimaryKey', SchemaEvolution: 'Auto', PersistSchema: ON, E1P: true, Serializer: 'StriimSerializer', CommitPolicy: 'EventCount=10000;Interval=15s', CheckpointTopicConfig: '{"PartitionCount":1,"ReplicationFactor":3}' ) FORMAT USING Global.AvroFormatter ( useSchemaRegistryConnectionProfile: 'true', SchemaRegistryConnectionProfileName: 'admin.ConfluentSchemaRegistry_Connection_Profile', formatAs: 'Native', SchemaRegistrySubjectName: 'UseTopicName' ) INPUT FROM STRIIM1; END APPLICATION demo;
In the DatabaseReader, set CreateSchema: true and QuiesceOnILCompletion: true so the initial-load-only run migrates the source schema before the CDC pipeline starts. In KafkaWriter, map each table to exactly one topic, set TopicKey to @metadata(TableName), and choose AutoCreateTopic, SchemaEvolution, and PersistSchema to match how you want the schema to be created and tracked going forward.
Using the Schema Conversion Tool (SCT)
The SCT generates initial Avro schemas from source database metadata, including:
Logical Avro data types for DML schema.
NOT NULL and PK constraints reflected as non-nullable fields.
All nullable fields assigned a default value of null.
Original source column name case preserved.
Detailed per-column metadata (PK info, data types, ordinal positions, scale, precision, nullability).
Example command:
bin/schemaConversionUtilityDev.sh -s=oracle -d="jdbc:oracle:thin:@//10.45.18.106:1521/orcl" -u="qatest" -p="***" -b='Sch.EMP;' -t="kafkanative"
The SCT outputs .json schema files that must be registered in the schema registry. See Registering an Externally Generated Schema for registration steps.
SCT command parameters: -s (source database type), -d (JDBC URL to the source database), -u (username), -p (password), -b (semicolon-separated list of source tables to generate schema for), -t (target type: kafkanative or kafkatable, matching the formatAs configuration).
Using External Tools
If creating schemas externally, follow these guidelines:
Schema structure must match the FormatAs setting in the AvroFormatter.
Condensed sample schema (Sch.EMP, Native vs. Table formatAs) — illustrating the structural difference:
Native (subject: Sch.EMP-value) — data/before are nested records; presence-bitmap records track which columns were present on each event:
{ "type": "record", "name": "EMP", "namespace": "Sch", "fields": [
{ "name": "data", "type": ["null", { "type": "record", "name": "data_record", "fields": [
{ "name": "EMP_ID", "type": {"type":"bytes","logicalType":"decimal","precision":38,"scale":0} },
{ "name": "FIRST_NAME", "type": ["null","string"], "default": null }
/* ... remaining columns follow the same pattern ... */
]}]},
{ "name": "before", "type": ["null", { "type": "record", "name": "before_record", "fields": [ /* same shape as data */ ] }] },
{ "name": "metadata", "type": ["null", {"type":"map","values":["null","string"]}] },
{ "name": "userdata", "type": ["null", {"type":"map","values":["null","string"]}] },
{ "name": "datapresenceinfo", "type": ["null", {"type":"record","name":"datapresenceinfo_record",
"fields": [{ "name": "EMP_ID", "type": "boolean" } /* one boolean field per column */ ]}] },
{ "name": "beforepresenceinfo", "type": ["null", { /* same shape as datapresenceinfo */ }] }
]}Table (subject: Sch.EMP-value) — only the data fields, flattened, no metadata/userdata/before/presence records:
{ "type": "record", "name": "EMP", "namespace": "Sch", "fields": [
{ "name": "EMP_ID", "type": {"type":"bytes","logicalType":"decimal","precision":38,"scale":0} },
{ "name": "FIRST_NAME", "type": ["null","string"], "default": null }
/* ... remaining columns follow the same pattern ... */
]}The full, uncondensed schema for all 8 Sch.EMP columns (plus the DDL schema and a sample DDL Kafka message) is available in the functional spec if an exact byte-for-byte reference is needed.
Field order must match the original source column ordinal order.
Field names must exactly match source schema column names, including case (Oracle/OJet: UPPERCASE; SQL Server/MySQL: preserved as created; PostgreSQL: unquoted = lowercase, quoted = exact case).
If a source column name contains special characters unsupported by Avro, the Avro field name may differ but an alias must preserve the original column name.
Registering an Externally Generated Schema
Register using curl. Match the subject name to the Subject Name Strategy configured in Kafka Writer.
Basic authentication:
curl -X POST \
-u <username>:<password> \
-H "Content-Type: application/vnd.schemaregistry.v1+json" \
--data "$(cat <jsonFilePath> | jq -c '.' | jq -Rs '{schema: .}')" \
https://<schemaRegistryHost>:<port>/subjects/<subjectName>/versionsConfluent Cloud API Key:
curl -X POST \
-u <apiKey>:<apiSecret> \
-H "Content-Type: application/vnd.schemaregistry.v1+json" \
--data "$(cat <jsonFilePath> | jq -c '.' | jq -Rs '{schema: .}')" \
https://<confluentCloudSchemaRegistryEndpoint>/subjects/<subjectName>/versionsMutual TLS:
curl -X POST \
--cert <clientCertificatePath> \
--key <clientKeyPath>:<keyPassword> \
--cacert <caCertificatePath> \
-H "Content-Type: application/vnd.schemaregistry.v1+json" \
--data "$(cat <jsonFilePath> | jq -c '.' | jq -Rs '{schema: .}')" \
https://<schemaRegistryHost>:<port>/subjects/<subjectName>/versionsNo authentication:
curl -X POST \
-H "Content-Type: application/vnd.schemaregistry.v1+json" \
--data "$(cat <jsonFilePath> | jq -c '.' | jq -Rs '{schema: .}')" \
http://<schemaRegistryHost>:<port>/subjects/<subjectName>/versionsInitial Schema Detection
When the first DML event for a mapped table arrives:
If a schema is present in the registry: the Avro record is constructed using the retrieved schema, avoiding discrepancies between typeUUID-generated and SCT-based schemas.
If no schema is present: typeUUID-based mapping is used. If CDDLAction is Process, there may be a mismatch in field-to-Avro-type mapping.
Schema Evolution
The SchemaEvolution property defines the action Kafka Writer takes upon encountering a DDL event. Applicable to all OLTP-based sources (Oracle, MySQL, SQL Server, PostgreSQL --- both IL and CDC).
Note
For CSV, JSON, and XML formatters, there is no schema tracking --- all incoming events are distributed based on the Topics and Partition Key settings regardless of DDL.
Schema Evolution: Auto
When a WAEvent with operation name DDL is received:
CREATE DDL: A new schema is created using the SCT, registered in the schema registry, and used as the initial schema.
ADD/DROP/MODIFY Column: A new schema is created on top of the initial schema and registered. DML events after the DDL use the latest evolved schema ID.
DDL messages are written to data topics if Persist DDL is set to TRUE.
DDL Persistence Rules
DDL messages are written to data topics only when Schema Evolution is Auto and Persist Schema is True (default).
If PartitionKey or TopicKey is absent in a DDL event, the application will HALT.
When Schema Evolution is Auto and Persist Schema is False, TopicKey must still be present --- otherwise the application will HALT.
When Schema Evolution is Manual, TopicKey must be present --- otherwise the application will HALT.
Schema Correlation During Restart
When an application restarts and replays past events, mismatches may occur between the incoming event structure and the latest schema. Kafka Writer ensures schema integrity via schema lineage tracked in the Striim MDR (InternalSchemaRegistry table). Upon restart, the appropriate DML schema for each incoming event is retrieved based on the event's position.
DDL Kafka Message Structure (Avro Formatter)
DDL messages contain:
Header: custom header if configured fields are present; otherwise empty.
Key (Custom): operation type or table name from user data or metadata.
Key (PrimaryKey): primary key column names with values set to NULL; if the DDL updates PK columns, message keys reflect the change.
Payload: DDL schema with DDL operations, source table information, source DDL string, DML Schema ID, and detailed column-level metadata.
DDL schemas are registered under subject name <Subject-Name>-DDLRecord. DDL Message Key schemas are registered under <subject-naming>-DDLKey.
Schema Evolution: Manual
When Schema Evolution is set to Manual, upon receiving a DDL event the adapter flushes all pending events and then halts gracefully.
Recovering After a Halt (Manual Schema Evolution)
Option 1 --- Switch to Auto:
Change the Schema Evolution setting to Auto.
Restart the application.
Option 2 --- Handle schema evolution externally:
Manually register the updated schema in the Schema Registry.
Create a CQ to skip the specific DDL event that caused the halt. Example:
SELECT i FROM inputstream i WHERE NOT ( TO_STRING(META(i, 'OperationType')) = 'DDL' AND TO_STRING(META(i, 'TableName')) = 'public.EMPLOYEE' AND TO_JSON_NODE(META(i, 'CDDLMetadata')).get('sql').asText() = 'alter table public."EMPLOYEE" add column company varchar(50)' );
Restart --- the CQ filters out the DDL event and Kafka Writer picks up the latest schema version from the registry.
Caution: Schema Evolution can remain set to Manual after applying this fix, so any future DDL still halts the application for controlled review. Also note: if a DDL event is skipped via CQ but the evolved schema is not manually applied to the target, the application HALTs with a Schema Mismatch Exception, or data loss can occur.
Schema Compatibility
Schema compatibility ensures older consumers can read new data and new consumers can read older data. Without compatibility rules, schema evolution can cause data corruption (deserialization failures), pipeline failure from 'poison pill' messages, and uncoordinated deployments.
Compatibility mode is set via the SchemaRegistryCompatibility property in the AvroFormatter. The default is None. All other compatibility modes are supported only with formatAs: Table and Native. Configuring a compatibility mode overrides any existing compatibility setting for the subject in the Schema Registry.
Note
Confluent Schema Registry's own default compatibility setting is Backward. Striim's default is None — change the setting explicitly if you rely on Confluent's default behavior. Compatibility checking functions correctly only when every schema for a table was generated through SCT, the schema conversion command-line utility, or an external process that accurately captures logical data types and NOT NULL constraints — manually registered schemas that don't follow this can silently break compatibility validation.
Mode | Description |
|---|---|
None | No compatibility checks are enforced; any schema change is allowed. |
Forward | Data written with the new schema can still be read using the previous schema version. |
Forward Transitive | The new schema is forward compatible with all previous schema versions, not just the latest. |
Backward | Data written with the old schema can still be read using the new schema. |
Backward Transitive | The new schema is backward compatible with all previous schema versions, not just the latest. |
Full | The new schema is both backward and forward compatible with the previous schema. |
Full Transitive | The new schema is both backward and forward compatible with all previous schema versions. |
The following DDL changes are allowed (✓) or blocked (✗) under each compatibility mode:
DDL Operation | None | Forward | Fwd. Transitive | Backward | Bwd. Transitive | Full | Full Transitive |
|---|---|---|---|---|---|---|---|
Create table (initial schema) | Yes | Yes | Yes | Yes | Yes | Yes | Yes |
Add nullable column | Yes | Yes | Yes | Yes | Yes | Yes | Yes |
Add non-nullable column | Yes | Yes | Yes | No | No | No | No |
Drop nullable column | Yes | Yes | Yes | Yes | Yes | Yes | Yes |
Drop non-nullable column | Yes | No | No | Yes | Yes | No | No |
Widen datatype of nullable column | Yes | No | No | Yes | Yes | No | No |
Narrow datatype of nullable column | Yes | Yes | Yes | No | No | No | No |
Widen datatype of non-nullable column | Yes | No | No | Yes | Yes | No | No |
Narrow datatype of non-nullable column | Yes | Yes | Yes | No | No | No | No |
Change column from nullable to non-nullable | Yes | Yes | Yes | No | No | No | No |
Connecting to Schema Registry
Setting up Connection to Confluent Schema Registry
Supported Authentication Mechanisms
Basic
Confluent Cloud API
Mutual TLS
None
Using SASL Authentication
Confluent Cloud API
Confluent Cloud API Keys are credentials used to authenticate clients connecting to Confluent Cloud services. Schema Registry in Confluent Cloud uses HTTPS and Basic Auth (API key/secret).
Prerequisites:
A Confluent Cloud Kafka Cluster.
An API Key and Secret. Create one via: Confluent Cloud Console > Clusters > Choose the Cluster > on the left, click Schema Registry > click API Access in the top-right > click Create Key.
Configure the Schema Registry Connection Profile:
Property Name | Value |
|---|---|
Schema Registry URL | Your Confluent Cloud Schema Registry endpoint |
Authentication Type | Confluent Cloud API |
API Key | Your Schema Registry API Key |
API Secret | Your Schema Registry API Secret |
Use SSL | True |
Basic Authentication
Basic authentication means clients authenticate with Schema Registry using a username and password, sent as a Base64-encoded string in the HTTP Authorization header.
Note
Basic authentication is not supported on the Confluent Cloud Schema Registry.
Configure the Schema Registry Connection Profile:
Property Name | Value |
|---|---|
Authentication Type | Basic |
Username | Your Schema Registry username |
Password | Your Schema Registry password |
Using Mutual TLS
Mutual TLS (mTLS) requires both the Schema Registry server and the client to present certificates. For steps to generate certificates, see Using Mutual TLS under Setting up Connection to Kafka.
For local (self-managed) Schema Registry deployment, configure the schema-registry.properties:
listeners=https://0.0.0.0:8081 ssl.client.auth=required ssl.keystore.location=/path/to/server.keystore.jks ssl.keystore.password=changeit ssl.key.password=changeit ssl.truststore.location=/path/to/server.truststore.jks ssl.truststore.password=changeit
Note
Mutual TLS is currently not supported for the Confluent Cloud Schema Registry.
Configure the Schema Registry Connection Profile with mTLS settings:
Property Name | Value |
|---|---|
Authentication Type | Mutual TLS |
Use SSL | True |
Use Certificate | True (recommended) or False |
CA Certificate | Path to CA PEM file (if Use Certificate is true) |
SSL Key Store Certificate Chain | PEM-formatted client certificate chain (if Use Certificate is true) |
SSL Key Store Key | PEM-formatted private key (if Use Certificate is true) |
SSL Key Password | Password if the private key is encrypted |
SSL Key Store Location | Path to keystore file (if Use Certificate is false) |
SSL Key Store Password | Keystore password (if Use Certificate is false) |
Setting up Connection to Karapace Schema Registry
Karapace is an open-source schema registry that is API-compatible with the Confluent Schema Registry API. It is commonly used with self-managed Apache Kafka deployments and with Aiven for Apache Kafka.
Important: Karapace is currently supported via direct URL configuration only. Kafka Writer does not support a Connection Profile for Karapace; instead, configure the schemaregistryurl and, if authentication is required, schemaregistryConfiguration properties directly in the AvroFormatter.
Registry | Connection Method | Supported Authentication |
|---|---|---|
Confluent Schema Registry | Connection Profile (recommended) or direct URL | Confluent Cloud API, Basic, Mutual TLS, None |
Karapace | Direct URL only (no Connection Profile) | Basic, None |
Example — Karapace with Basic authentication:
FORMAT USING Global.AvroFormatter ( schemaregistryurl: 'https://karapace.example.com:8081', schemaregistryConfiguration: 'basic.auth.user.info=username:password,basic.auth.credentials.source=USER_INFO', formatAs: 'Native', SchemaRegistrySubjectName: 'UseTopicName' )
Example — Aiven for Apache Kafka (Karapace):
FORMAT USING Global.AvroFormatter ( schemaregistryurl: 'https://kafka-xxxxx.aivencloud.com:28139', schemaregistryConfiguration: 'basic.auth.user.info=avnadmin:your-password,basic.auth.credentials.source=USER_INFO', formatAs: 'Native', SchemaRegistrySubjectName: 'UseTopicName' )
DSV Formatter
The DSV (Delimiter-Separated Values) formatter is supported for WAEvent input streams.
The DSV Formatter used by Kafka Writer is unchanged from previous Striim releases; see the DSV Formatter documentation for full delimiter, field-ordering, and quoting details.
Example DSV Formatter configuration used with Kafka Writer:
FORMAT USING Global.DSVFormatter ( quotecharacter: '\"', columndelimiter: ',', nullvalue: 'NULL', usequotes: 'false', rowdelimiter: '\n', standard: 'none', header: 'false' )
JSON Formatter
The JSON Formatter is supported for JsonNodeEvent and WAEvent input streams.
The JSON Formatter used by Kafka Writer is unchanged from previous Striim releases; see the JSON Formatter documentation for configuration details and examples.