Kafka Writer Operational Considerations
Monitoring and Metrics
The Kafka Writer publishes the following metrics for monitoring:
Metric Name | Description |
|---|---|
No. of Kafka Messages Sent | Total number of messages successfully published to Kafka across all topics and partitions since application start. |
Kafka Messages Rate | Number of messages published to Kafka per second. |
Kafka Write Rate | Average number of bytes sent per second to Kafka. |
Discarded Event Count | Total number of events discarded by the adapter. |
Buffered Messages Size (bytes) | Current memory usage for buffered messages or events in the adapter. |
Kafka Producer ID(s) | ID of the Kafka Producer instance used for publishing messages. |
Average Latency | Average time (ms) for Kafka request and acknowledgment per non-transactional buffer or transaction. |
Maximum Latency | Maximum time (ms) for Kafka request and acknowledgment per non-transactional buffer or transaction. |
Last Send Timestamp | Timestamp of the most recent successful Send operation to Kafka. |
No. of Retries | Total number of retry attempts made by the adapter due to connection or processing failures. |
Last Retry Attempt Timestamp | Timestamp of the most recent retry attempt. |
The following transactional metrics are visible only when E1P is enabled:
Metric Name | Description |
|---|---|
Successful Transactions | Total number of committed transactions executed. |
Average Messages per Transaction | Average number of events in each committed transaction. |
Average Transaction Size (bytes) | Average size of committed transactions in bytes. |
Max Transaction Size (bytes) | Maximum size in bytes of any committed transaction since application start. |
Average Transaction Accumulation Time (ms) | Average time spent accumulating events for a transaction. |
Commit Lag (ms) per Adapter Instance | Time taken between execution of a batch and its publication from each source, in milliseconds. |
Kafka Transactional ID(s) | ID of the Kafka Transactional Producer instance used for publishing messages. |
Per-topic and per-partition metrics are also available. For each topic, the adapter tracks: Mapped Source Entity, Total Kafka Messages Sent, Total DDLs Sent, Created By (if created during current run). For each partition within a topic: Latest Offset, Total No. of Messages, No. of DDLs, Avg Message Size, Last Activity Timestamp.
A separate Last Transaction Information metric (JSON format, visible only when E1P is enabled) reports details of the most recently committed transaction per adapter instance: No. of Events in the Transaction, Transaction Size (bytes), Transaction Time (ms), Transaction Accumulation Time (ms), and Topic Partition Count, along with its own nested per-topic and per-partition breakdown scoped to that specific transaction.
DDL Information metrics are also tracked per adapter instance: Schema Evolution (configured action: Auto or Manual), No. of DDLs Handled.
Troubleshooting
Known Limitations
Case Sensitivity
Case sensitivity is supported for topic names (see Topic Configuration) across all formatters, with the exception of the Avro formatter.
Handling CDC Events from MSJet
When using MSJet with KafkaWriter:
During Initial Load (IL), topics are created using a three-part name (database.schema.table).
During CDC processing, topics are created using a two-part name (schema.table).
This inconsistency may result in IL and CDC applications writing to different topics for the same table and can lead to schema registry conflicts when subject names are derived from topic names.
Additional known limitations
FileReader with subdirectories: when using FileReader as the source with includeSubDirectories set to true and TopicKey set to @metadata(FileName), the FileName metadata value contains the full file path. Since the path includes directory separators, the resulting value does not comply with Kafka topic name rules. To avoid this, either disable includeSubDirectories on the source or use a different field for TopicKey.
Dropping NOT NULL or primary key constraints: when SchemaEvolution is set to Auto and the Avro Formatter's formatAs is set to Native or Table, dropping a NOT NULL or primary key constraint via DDL is not supported.
DDL column names with unsupported special characters: when SchemaEvolution is set to Auto and PersistSchema is On, the application HALTs if a DDL event contains column names with special characters that Avro doesn't support.
Workarounds:
Query + Tables combination --- use a three-part name in the query with a two-part name in the tables property. (Limitation: only one table can be configured per adapter).
Custom CQ --- apply a continuous query in the CDC application to enforce three-part naming in the metadata.
Prefixed wildcards --- use patterns such as %,<database>.% instead of full wildcards.
Pre-created topics --- disable AutoCreateTopic and configure explicitly created topics with direct mapping and SchemaRegistrySubjectName = UseTopicName.
Note
For scenarios involving multiple databases, create separate applications for each database rather than attaching multiple source adapters to a single target application.
Configuration Issues
Problem | Cause | Resolution |
|---|---|---|
Failed to parse Connection Retry Policy: Invalid key '<Key>' | The Connection Retry Policy contains an unrecognized key. | Use only valid keys: initialRetryDelay, retryDelayMultiplier, maxRetryDelay, maxAttempts, totalTimeout. Format: Key=Value;Key=Value. |
Connection Retry Policy '<key>' must be a valid time value | Value for a time-based parameter is not in valid format. | Use numeric milliseconds (5000) or time units (5s, 5m, 1h). No spaces between number and unit. |
Provide the Topics in this format: source,TopicName; | Topics property syntax is incorrect. | Single topic: use topic name only. Multiple: 'source,TopicName;source2,TopicName2'. Semicolons separate mappings; comma separates source from topic. |
Topic Key is not supported with single topic name | TopicKey set but Topics contains only a single topic name. | Remove TopicKey when using a single topic. Use topic mapping format if routing is needed. |
Topic Key is mandatory when using multiple topics | Topics has multiple mappings but TopicKey is not specified. | Set TopicKey to specify the routing field. Examples: @metadata(TableName) for CDC, field name for typed events. |
DataTopicConfig must be in valid JSON format | DataTopicConfig contains invalid JSON syntax. | Ensure valid JSON with double quotes for keys/strings. Example: {"PartitionCount":1,"ReplicationFactor":1}. |
CommitPolicy format invalid | CommitPolicy is not in EventCount=<n>;Interval=<duration> format. | Example: EventCount=1000;Interval=5m. Semicolon separates parameters; = separates key from value. |
Custom Partition Key must contain non-empty values | CustomPartitionKey has empty values between semicolons. | Use format: field1;field2;field3. Remove trailing/leading semicolons and empty values. |
Custom Message Key cannot be empty when MessageKey=Custom | MessageKey is Custom but CustomMessageKey is empty. | Configure CustomMessageKey with key=value pairs. Example: id=customerId;name=customerName. |
Timed out waiting for node assignment. Call: listTopics | Kafka AdminClient cannot connect to broker cluster. Broker addresses incorrect, network latency, firewall blocking, or brokers down. | Verify Broker Address is correct and reachable. Ensure firewall allows outbound connections to Kafka ports. Confirm brokers are running. |
Unexpected authentication error | Schema Registry URL is unreachable (wrong host/port, HTTP vs HTTPS mismatch, DNS/firewall/VPC). | Ensure the URL in the Connection Profile is valid and reachable. Verify the Schema Registry is running. |
Metadata/Userdata field not found: key <keyName> | A @metadata() or @userdata() key in TopicKey, PartitionKey, MessageKey, or MessageHeader does not exist in the incoming event. | Verify the field exists in the WAEvent metadata or userdata. Check spelling (case-sensitive). |
Topic <TopicName> does not exist and auto-create is disabled | The topic does not exist and AutoCreateTopic is false. Or user lacks DESCRIBE permission. | Set AutoCreateTopic=true, or manually create the topic. Ensure DESCRIBE ACL is granted on the topic. |
Cluster authorization failed | Kafka user lacks DESCRIBE_CONFIGS ACL required for auto topic creation with default settings. | Pre-create topics and disable AutoCreateTopic, or configure PartitionCount and ReplicationFactor explicitly in DataTopicConfig. |
Authorization failed | The authenticated Kafka user does not have all required ACLs. | Grant all required ACLs per the Required ACLs table. |
Connection Retry Policy '<key>' must be a valid number | Value for retryDelayMultiplier or maxAttempts is not numeric. | Provide a valid numeric value, digits only. Example: maxAttempts=5, retryDelayMultiplier=2. |
Connection Retry Policy '<key>' must be a non-negative/positive value | Time parameters need ≥ 0; numeric parameters (retryDelayMultiplier, maxAttempts) need > 0. | Use non-negative values for time parameters and positive values for numeric parameters. Check for stray negative signs or zeros. |
Invalid DataTopicConfig properties: <key> | DataTopicConfig contains a property name that isn't a recognized Kafka topic configuration. | Use only supported properties: PartitionCount, ReplicationFactor, MaxMessageSize, RetentionSize, RetentionTime, CleanUpPolicy. |
Message Key must be provided if cleanup.policy is set to compact | DataTopicConfig sets cleanup.policy to compact but MessageKey is not configured. | Set MessageKey to PrimaryKey or Custom, or change cleanup.policy to delete if compaction isn't needed. |
Data topic name '<topicName>' contains wildcard character (%) which is only allowed with topic mapping | The Topics property has a % character without mapping syntax. | Use topic mapping format, e.g. 'source,TopicName%', or remove the % if wildcarding isn't needed. |
Data topic name '<topicName>' is not valid | The resolved topic name contains characters Kafka doesn't allow. | Use only alphanumeric characters, dot (.), underscore (_), and hyphen (-); use explicit mapping for source names with unsupported characters. |
Data topic name '<topicName>' exceeds the length limit of 249 characters | The resolved topic name is longer than Kafka's 249-character limit. | Shorten the topic name via explicit mapping, or remove long wildcard prefixes/suffixes. |
Topic replication factor must be 3 | Confluent Cloud enforces a fixed replication factor of 3. | Remove the replication.factor override from TopicConfig, or set it explicitly to 3. |
Custom Message Key must be in 'key=value;key2=value2' format | CustomMessageKey isn't in valid key=value pair format. | Use key=value pairs separated by semicolons, e.g. id=customerId;region=state. |
Message Header must be in 'key=value;key2=value2' format | MessageHeader isn't in valid key=value pair format. | Use key=value pairs separated by semicolons, e.g. source=application;table=@metadata(TableName). |
Invalid SchemaEvolution value | SchemaEvolution is set to a value other than Auto or Manual. | Use only Auto or Manual; check for typos. |
Problem getting token <name> from event data fields | A field referenced in TopicKey, PartitionKey, MessageKey, or MessageHeader doesn't exist in the event's type definition. | Verify the field exists (case-sensitive) using DESCRIBE TYPE. |
Retries and Error Recovery
See also Handling Retries for the underlying retry configuration. When E1P is on, exactly-once guarantees apply to both internal and external retries --- no duplicates will be produced.
With A1P:
Internal Kafka retries will not cause duplication or out-of-order messages if the producer is configured with max.in.flight.requests.per.connection ≤ 5 and enable.idempotence=true.
External (writer-level) retries create a new Kafka producer and retry all messages --- duplicates can occur on connection glitches or changes in connection information.
Handling Special Characters
Kafka topics created via Kafka Writer support dot (.), underscore (_), and hyphen (-).
Avro schemas registered via the Avro Formatter support: underscores and dots in namespace; underscores in Avro record names and field names.
In the legacy Kafka Writer (pre-5.4), special characters were converted to a mangled name format (_<Name of special character>_). The new Kafka Writer will HALT on unsupported special characters. Backward compatibility mode retains the mangled naming convention.
Avro naming limitation: The Avro Formatter currently uses Avro 1.7.7, which does not support special characters in Avro record names or namespaces. Only letters, numbers, and underscores are allowed, and the name must start with a letter or underscore.
Altering Application
The table below summarizes which Kafka Writer properties can be changed on a running application via ALTER, and what to expect afterward. Properties marked "Not recommended" or requiring a quiesce should only be changed with the application quiesced first, to avoid data loss or inconsistent distribution.
Property | Can be altered? | Expected changes after ALTER |
|---|---|---|
ConnectionProfileName | Yes | The application uses the newly attached connection profile. |
ConnectionRetryPolicy | Yes | The altered retry policy applies on the next connection failure. |
Topics | Not recommended (quiesce first) | Subsequent events are mapped based on the newly configured Topics. |
TopicKey | Not recommended (quiesce first) | Make sure Topics matches the new TopicKey before restarting; subsequent events are mapped accordingly. |
AutoCreateTopic | Yes | If changed true→false, topics must already exist. If changed false→true, new topics are auto-created. |
DataTopicConfig | Yes | Only affects topics created after the ALTER. |
PartitionKey / CustomPartitionKey | Not recommended (quiesce first) | Partition distribution changes for subsequent events. |
E1P | Yes (quiesce first) | Toggling requires Recovery to be on; switches between exactly-once and at-least-once semantics. |
CheckpointTopic | Not recommended (quiesce first) | Checkpointing moves to the newly configured topic; may cause data loss if not quiesced properly. |
CheckpointTopicConfig | Not recommended | Only takes effect if the checkpoint topic is (re)created after the ALTER. |
CommitPolicy | Yes | The new commit policy applies to subsequent events. |
ParallelThreads | Yes (quiesce first) | With E1P: decreasing is safe (extra checkpoint partitions go unused); increasing requires manually growing the checkpoint topic's partition count first, or the application HALTs. |
MessageHeader | Yes (quiesce first) | New headers apply to subsequent messages only; existing messages retain old headers. |
MessageKey / CustomMessageKey | Yes (quiesce first) | Affects message keys, and partition distribution if partitioning depends on the message key. |
Serializer | Not recommended (quiesce first) | Serialization format changes; consumers may need updating, or use a new topic. |
SchemaEvolution | Yes (quiesce first) | Changes DDL handling behavior for subsequent schema changes. |
PersistSchema | Yes (quiesce first) | If changed to false, DDL events stop being persisted to the data topic. |
Additionally: switching the Avro Formatter's formatAs from Table or Native to Default is supported while altering; switching from Default to Table or Native (or between Native and Table) is not supported and requires dropping and recreating the application with a different subject name.