The following provides usage information for the Apache Kafka® SMT
Set the schema name, version or both on the record’s key (
org.apache.kafka.connect.transforms.SetSchemaMetadata$Key) or value (
This SMT can be used to set the schema name, version, or both on Connect records. Since the schema name includes the namespace, a common SetSchemaMetadata SMT use case is to change the name and namespace of the schemas used in Apache Kafka® record keys and values.
Source connectors generate records with schemas defined by the connector, and often generate the names and versions of key and value schemas based upon source-specific information. For example, a database source connector might use the name of the table from which rows are read in the schema names. If these generated schema names do not adhere to the naming convention you need, you can use this SMT to override the generated names of the key and/or value schemas in the source records produced by the source connector.
Sink connectors that consume records from Kafka topics may the names of the schemas to indicate how the connector maps the Kafka records into the external system. If the names of the schemas used in the record keys and values don’t result in the desired mapping, you can use this SMT to change the name of the key and/or value schemas in the consumed source records, before those records are processed by the sink connector.
This configuration snippet shows how to use
SetSchemaMetadata to set a
schema name and version.
"transforms": "SetSchemaMetadata", "transforms.SetSchemaMetadata.type": "org.apache.kafka.connect.transforms.SetSchemaMetadata$Value", "transforms.SetSchemaMetadata.schema.name": "order-value" "transforms.SetSchemaMetadata.schema.version": "2"
Example: See https://docs.confluent.io/current/tutorials/examples/connect-streams-pipeline/docs/index.html#example-3-jdbc-source-connector-with-specificavro-key-string-null-and-value-specificavro. The example uses
SetSchemaMetadata to allow a connector to set the namespace in the schema.
||Schema name to set.||string||
||Schema version to set.||int||
Transformations can be configured with predicates so that the transformation is applied only to records which satisfy a condition. You can use predicates in a transformation chain and, when combined with the Filter (Apache Kafka), predicates can conditionally filter out specific records. For details and examples, see Predicates.