How to replace a field value using Kafka Transform ReplaceField in Sink Connector

Viewed 29

I want to rename the value of key name "id" I've use case where producer publish messages to this topic (product-topic) either of these messages "id": "test.product.mobile" or "id": "test.product.computer"

My HTTP sink connector consume the message from this topic and want to do the transformation (rename the field's value)

For example,

  • if producer sends "id": "test.product.mobile" I want to replace like this "id": "test.product.iPhone"
  • if producer sends "id": "test.product.computer" I want to replace like this "id": "test.product.mac"

I'm using HTTP sink connector and transform package to replace field value, but it's not working as expected. Please find the connector configuration below:

{
    "connector.class": "io.confluent.connect.http.HttpSinkConnector",
    "confluent.topic.bootstrap.servers": "localhost:9092",
    "topics": "product-topic",
    "tasks.max": "1",
    "http.api.url": "http://localhost:8080/product/create",
    "reporter.bootstrap.servers": "localhost:9092",
    "reporter.error.topic.name": "error-responses",
    "reporter.result.topic.name": "success-responses",
    "reporter.error.topic.replication.factor": "1",
    "confluent.topic.replication.factor": "1",
    "errors.tolerance": "all",
    "value.converter.schemas.enable": "false",
    "batch.json.as.array": "true",
    "name": "Product-Connector",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "reporter.result.topic.replication.factor": "1",
    "transforms": "RenameField",
    "transforms.RenameField.type": "org.apache.kafka.connect.transforms.ReplaceField$Key",
    "transforms.RenameField.renames": "id:test.product.iPhone"
}

Producer send messages like below

{
    "id": "test.product.mobile",
    "price": "1232"
}

{
    "id": "test.product.computer",
    "price": "2032"
} 

Expected Output:

{
    "id": "test.product.iPhone",
    "price": "1232"
}

{
    "id": "test.product.mac",
    "price": "2032"
} 

I referred the Kafka Confluent Docs to rename a field but that example works well if we want to replace the key name but not value. Can someone please help me with use case - what needs to be change to rename the field value?

Appreciated your help in advance. Thanks!

1 Answers

It's not possible to replace field value text with any (included) SMTs, outside of masking, no.

You could write (or find) your own SMT, but otherwise, the recommended pattern for this is a KStreams/ksqlDB process.

Or, simply have your initial Kafka producer send the values that you want to sink to the HTTP server.

Related