Create Kafka Writer Application
Kafka Writer (the new KafkaWriter target described in this guide) accepts the same input stream types supported by its formatters: WAEvent (from OLTP, DWH, file, message bus, and application sources), AvroEvent, JSONNodeEvent, ParquetEvent, and TypedEvent.
Prerequisites
Before creating a Kafka Writer application, ensure the following are in place:
Running Striim 5.4.2 or later --- KafkaWriter as described in this guide is available starting in this release.
A Kafka Connection Profile has been created and tested (see Setting up Connection to Kafka).
If using a Schema Registry, a Schema Registry Connection Profile has been created (see Connecting to Schema Registry).
If using the Striim Vault for certificates, the vault entries are populated (see Using the Striim Vault).
The target Kafka cluster is accessible from the Striim node.
Required ACLs are granted to the Striim service account.
Your source produces one of the supported input stream types listed above.
Topic Configuration
Topics Strategy
Data can be routed to a single topic or distributed across multiple topics based on source table names, field values, or metadata. The Topics property accepts:
A single topic name (e.g., mytopic) --- all events go to one topic.
A wildcard mapping using % to auto-name topics from source entity names.
An explicit semicolon-separated mapping of source entities to topic names.
Topic Creation
Topics can be pre-created in Kafka, or created automatically by the Kafka Writer if AutoCreateTopic is set to true. When auto-creating, use CheckpointTopicConfig to specify replication factor and partition count for the checkpointing topic.
Data Topics
Data topics hold the actual Kafka messages produced from source events. Topic names are determined by the Topics property and the TopicKey field (for multi-topic scenarios).
Wildcard Topic Mapping
Use % as a wildcard to name topics dynamically from the source entity name. For example:
Topics: %,% with TopicKey: @metadata(TableName) --- creates one topic per source table, named after the table (e.g., src.EMP, src.DEPT).
Topics: src.%,% or %,striim_% --- add a prefix or suffix to each auto-named topic.
Explicit Topic Mapping
Map specific source entities to named topics:
Topics: EMP,employee-topic; src.DEPT,dept-topic; src.CUSTOMER,customer-topic TopicKey: @metadata(TableName)
Checkpoint Topics
When E1P is enabled, Kafka Writer uses an additional checkpointing topic to store recovery information. If this topic does not exist, the writer creates it automatically (requires topic creation permissions on the Kafka cluster). The checkpointing topic configuration can be customized via the CheckpointTopicConfig property.
Discarded Events
Events are discarded and not written to Kafka in the following scenarios:
Topic Mapping Failure --- the event cannot be mapped to any configured Kafka topic.
Missing Topic Key Value --- the configured topic key field is present but has a null or empty value.
The total count of discarded events can be monitored using the Discarded Event Count metric.
Best Practices
DataTopicConfig
By default, data topics are created using the broker's default topic configuration settings. To override these defaults, set the DataTopicConfig property as a JSON string.
Recommended: Increase the ReplicationFactor for higher availability. Example:
{"PartitionCount":1,"ReplicationFactor":3}CheckpointTopicConfig
The checkpoint topic has the following default configuration:
{"PartitionCount":1,"ReplicationFactor":3,"CleanUpPolicy":"compact","min.cleanable.dirty.ratio":"0.5","segment.ms":"86400000","segment.bytes":"1073741824","min.compaction.lag.ms":"3600000","max.compaction.lag.ms":"604800000"}Recommended settings:
Partition Count: 1 if parallel threads are not configured. If parallel threads are configured, set the partition count equal to the number of parallel threads.
Replication Factor: Higher value (e.g., 3 or more) for high availability and fault tolerance of checkpoint data.
Cleanup Policy (cleanup.policy): Must be set to compact. Do not change unless explicitly advised by Striim Support.
Compaction Frequency (min.cleanable.dirty.ratio): 0.5 --- compact segments once 50% of records are dirty. Helps reduce disk usage without overloading the CPU.
Compaction Timing: min.compaction.lag.ms = 3,600,000 ms (1 hour); max.compaction.lag.ms = 604,800,000 ms (7 days). Prevents very recent records from being compacted immediately while ensuring older records are compacted regularly.
Log Segment Settings: segment.ms = 86,400,000 ms (1 day); segment.bytes = 1,073,741,824 bytes (1 GB). Ensures log segments are rolled more frequently, making them eligible for compaction sooner.
Select Partition Strategy for Each Topic
Topics and Partition Strategy
Kafka messages are distributed across partitions within a topic based on the Partition Key. Striim supports four combinations:
Single topic, one partition: All source data goes to one target topic, partition 0. Preferred when preserving the order of source DML operations.
Single topic, multiple partitions: All source data goes to one topic but is distributed across partitions. Order is preserved only within a single partition.
Multiple topics, single partition each: Wildcard or explicit mapping of source entities to topics; data is written to partition 0 of each topic.
Multiple topics, multiple partitions: Extension of the previous scenario --- source data is routed to multiple topics and further partitioned within each topic.
Configuring Partitioning
Partition Key
The Partition Key determines how events are distributed across partitions.
None: All messages are routed to partition 0. Use when ordering across all events is required.
Custom: Specify a field name or metadata reference as the partition key. Examples:
Field-based: PartitionKey = Custom, CustomPartitionKey = deptId
Metadata-based (OLTP CDC): CustomPartitionKey = @metadata(OperationName)
File-based: CustomPartitionKey = @metadata(directory)
UseMessageKey: The Message Key is also used as the Partition Key. This is useful when message ordering per entity is important.
Note
The specified partition key must be present in all incoming events (except control and DDL events when Persist Schema is OFF). If absent, the application will HALT.
Partitioning Method
Default --- Hash-Based Partitioning
Striim uses a hash-based algorithm by default:
Partition = hash(partition_key) % number_of_partitions
This guarantees that the same partition key always maps to the same partition, which is required for maintaining event ordering per entity.
Note
If the number of partitions in a topic is increased, partition assignments for existing keys will change. Always quiesce the application before altering the number of Kafka partitions.
Hash Collisions: If two different partition keys produce the same hash value, both keys are assigned to the same partition.
Custom Partitioning
The Kafka Writer supports custom partitioners via the Partitioner.class property in the Producer Configurations of the Kafka Connection Profile. Custom partitioners must implement the PartitionerIntf interface.
Important constraints:
One partitioning strategy per target --- the Kafka Writer applies the same partitioning logic to all topics within a single target.
For E1P, the partitioner must always return the same partition ID for a given partition key, even during retries.
Example RangePartitioner implementation:
package com.example.kafka; import com.striim.custom.partitioner.PartitionerIntf; import org.apache.kafka.common.Cluster; import java.util.List; import java.util.Map; public class RangePartitioner implements PartitionerIntf { private int rangeSize = 100; @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { List<?> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (key == null) { return 0; } int keyInt; try { keyInt = Integer.parseInt(key.toString()); } catch (NumberFormatException e) { return 0; } return (keyInt / rangeSize) % numPartitions; } @Override public void close() {} }
Setting up Kafka Message Structure
Each incoming event is converted into a Kafka message (record) consisting of an optional header, a key, and a value (payload).
Message Header
See also Headers and Metadata for the conceptual overview. By default, messages are published without a Kafka header. You can define dynamic or static headers:
Dynamic header: reference fields from payload, metadata, or user data (only when the incoming stream contains typed events).
Static header: a fixed key-value pair applied to every message.
Multiple headers: use the UI widget or provide a semicolon-separated value in TQL.
Header format: key=value pairs. Example:
MessageHeader: 'TableName=@MetaData(TableName)'
Supported event types: JSONNodeEvent, WAEvent, AvroEvent, TypedEvent (all formatters).
Message Key
Three configurations are available for the Kafka message key:
None: Messages are published without a Kafka message key. This is the default. Applies to all event types.
Custom: Define a dynamic message key by referencing fields from payload, metadata, or user data, or add a static key. Applies to all event types.
PrimaryKey: The message key is automatically constructed from primary key column values of the source table (OLTP or DWH sources only). If no primary key columns are defined, all columns are used. Changes to PK definitions due to DDL are reflected in message keys when SchemaEvolution is set to Auto.
Serialization
Message Keys can be serialized using one of the following formatters: JSON, Avro, DSV, XML.
When using the Avro Formatter, message keys are Avro records with key-value pair fields. The field type is based on the incoming event field's type for OLTP sources; for non-OLTP sources, the type is String. A separate schema is registered using the subject naming strategy with a -key suffix.
Mapping Partition Keys to Kafka Message Key
When PartitionKey is set to UseMessageKey, the Message Key value is also used to determine the target partition. This ensures that all events with the same message key always land in the same partition --- a common pattern for maintaining per-entity ordering.
Examples:
Scenario 1 (OLTP CDC):
MessageKey: Custom, CustomMessageKey: @metadata(TableName),@metadata(OperationName), PartitionKey: UseMessageKey
Scenario 2 (File-based):
MessageKey: Custom, CustomMessageKey: @metadata(directory);@metadata(FileName), PartitionKey: UseMessageKey
Scenario 3 (Primary Key):
MessageKey: PrimaryKey, PartitionKey: UseMessageKey --- distribution is based on primary key columns.
Message Payload
Each incoming event's contents (data, metadata, user data, and other fields depending on the Members configuration) are formatted and added to the Kafka message value. By default Striim uses the Striim Serializer. For Avro Formatters, this can be set to Confluent Serializer via the Serializer property.
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 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 enforced when the schema evolves. See Schema Compatibility below for all supported modes. |
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.
Select Delivery Semantics
E1P --- Exactly-Once Processing
E1P is the default setting (E1P: true). In this mode:
Kafka transactions are used to ensure exactly-once delivery, avoiding duplicates even on retries.
A checkpointing topic is required. The writer creates it automatically if it does not exist (requires topic creation permission).
The transaction batch size is governed by the Commit Policy (see Kafka Writer Properties).
Required producer settings (set automatically): enable.idempotence=true and acks=all.
A1P --- At-Least-Once Processing
Set E1P: false to use at-least-once semantics. In this mode:
No checkpointing topic is required.
Duplicates may occur during retries --- acceptable if the downstream consumer is idempotent.
For A1P without duplicates on internal Kafka retries, configure the producer with: max.in.flight.requests.per.connection ≤ 5 and enable.idempotence=true.
Handling Retries
See also Retry and Recovery for a high-level overview. Two levels of retry occur:
Internal retry --- managed by the Kafka Producer Client. With E1P enabled, idempotent producers ensure no duplicates or out-of-order messages.
Writer-level retry --- Striim creates a new Kafka producer and retries all pending messages. This handles credential rotation and retriable exception codes. With A1P, duplicates can occur on connection glitches or connection information changes.
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.
Using External Tools
If creating schemas externally, follow these guidelines:
Schema structure must match the FormatAs setting in the AvroFormatter.
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.
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.
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 |
Handling Transactions (Begin/Commit CDC Events)
For WAEvent originating from OLTP sources: if Filter Source Transaction Boundary is enabled at the source, the target will not receive any control events.
Distribution
All events, including Begin/Commit events, are sent to partition 0 if no partition key is specified.
If incoming events are distributed using metadata or user data keys, Begin/Commit events cannot be distributed if they do not have the configured keys --- they will be discarded.
Primary Key partitioning: Begin/Commit events do not have table-level or primary key information and will be discarded.
Schema Tracking
If the Avro Formatter is used, a separate schema for control events is created and registered under the subject name: <subject-name>_ControlRecord.
Message Structure for Begin/Commit Events
Header: custom header added if required metadata/userdata is present; otherwise empty.
Key (Custom): operation type or table name is set as the message key if present in user data or metadata.
Key (PrimaryKey): control events do not have table-level information, so the message key will be empty.
In formatAs: Table mode, control events are skipped.
Sample Messages
Begin event (raw WAEvent):
{
"pos": null,
"_id": null,
"timeStamp": 1753773949048,
"originTimeStamp": 1753773949048,
"key": null,
"data": ["waction", "BEGIN"],
"metadata": {
"BinlogFile": "ON.000039",
"TxnID": "1:000039:2533:1753773949000",
"OperationName": "BEGIN",
"TimeStamp": 1753773949000,
"BinlogPosition": 2533
}
}Begin/Commit event --- FormatAs: Default:
{
"metadata": {
"BinlogFile": "ON.000039",
"TxnID": "1:000039:2897:1753774226000",
"OperationName": "BEGIN",
"TimeStamp": "2025-07-29T00:30:26.000-07:00",
"BinlogPosition": "2897"
},
"data": {"0": "waction", "1": "BEGIN"},
"before": null,
"userdata": null
}