Sink Connector Post Processors

Create a Custom Document Id Strategy

If the built-in document id adder strategies do not cover your use case, you can define a custom document id strategy by following the steps below:

1. Create a Java class that implements the interface IdStrategy and contains your custom configuration logic.

2. Compile the class to a JAR file.

3. Add the compiled JAR to the class path / plugin path for all your Kafka workers. For more information about plugin paths, see the Confluent documentation.

4. Update the document.id.strategy setting to the full class name of your custom class in all your Kafka workers.

For example implementations of the IdStrategy interface, see the source code directory that contains id strategy implementations packaged with the connector.

Allow List Wildcard Examples

You can use the * wildcard to match multiple field names. The following example configuration matches the following fields:

• The top-level field named "city"

• The fields named "average" that are subdocuments of any top-level field that starts with the name "wind_speed".

post.processor.chain=com.mongodb.kafka.connect.sink.processor.AllowListValueProjector
value.projection.type=AllowList
value.projection.list=city,wind_speed*.average

After the post processor applies the allow list projection, it outputs the following record:

{
  "city": "Springfield",
  "wind_speed_10m": {
    "average": 3,
  },
  "wind_speed_80m": {
    "average": 8,
  }
}

You can use the ** wildcard which matches objects at any level starting from the one at which you specify the wildcard. The following wildcard matching example projects any document that contains the field named "low".

post.processor.chain=com.mongodb.kafka.connect.sink.processor.AllowListValueProjector
value.projection.type=AllowList
value.projection.list=**.low

The post processor that applies the projection outputs the following record:

{
  "temperature": {
    "high": 28,
    "low": 24,
    "units": "C"
  },
  "soil_conditions": {
    "temperature": {
      "high": 22,
      "low": 17,
      "units": "C"
    }
  }
}

Block List Wildcard Example

You can use the wildcard patterns to match fields at a specific document level as shown in the following block list configuration example:

post.processor.chain=com.mongodb.kafka.connect.sink.processor.BlockListValueProjector
value.projection.type=BlockList
value.projection.list=*.*.temperature

{
  "city": "Springfield",
  "temperature": {
    "high": 28,
    "low": 24,
    "units": "C"
  },
  "wind_speed_10m": {
    "average": 3,
    "units": "km/h"
  },
  "wind_speed_80m": {
    "average": 8,
    "units": "km/h"
  },
  "soil_conditions": {
    "moisture": {
      "average": 340,
      "units": "mm"
    }
  }
}

Rename by Mapping Example

The RenameByMapping post processor setting specifies one or more JSON objects that assign fields matching a string to a new name. Each object contains the text to match in the oldName element and the replacement text in the newName element as described in the table below.

The following example property matches the "location" field of a key document and renames it to "country":

field.renamer.mapping=[{"oldName":"key.location", "newName":"country"}]

This setting instructs the RenameByMapping post processor to transform the original key document to the following document:

{
  "country": "Provence",
  "date_month": "October",
  "date_day": 17
}

You can perform a similar field name assignment for value documents by specifying the value document with the appended field name in the oldName field as follows:

field.renamer.mapping=[{"oldName":"value.flapjacks", "newName":"crepes"}]

This setting instructs the RenameByMapping post processor to transform the original value document to the following document:

{
  "crepes": {
    "purchased": 598,
    "size": "large"
  }
}

You can also specify one or more mappings in the field.renamer.mapping property by using a JSON array in string format as shown in the following setting:

field.renamer.mapping=[{ "oldName":"key.location", "newName":"city" }, { "oldName":"value.crepes", "newName":"flapjacks" }]

Rename by Regular Expression

The RenameByRegex post processor setting specifies the field names and text patterns that it should match, and replacement values for the matched text. You can specify one or more renaming expressions in JSON objects containing the fields described in the following table:

The following example setting instructs the post processor to perform the following:

• Match any field names in the key document that start with "date". In the set of matching fields, replace all text that matches the pattern _ with the - character.

• Match any field names in the value document that are subdocuments of crepes. In the set of matching fields, replace all text that matches the pattern purchased with quantity.

field.renamer.regexp=[{"regexp":"^key\\.date.*$","pattern":"_","replace":"-"}, {"regexp":"^value\\.crepes\\..*","pattern":"purchased","replace":"quantity"}]

When the connector applies the post processor to the example key document and the example value document, it outputs the following:

Key Document

{
  "location": "Provence",
  "date-month": "October",
  "date-day": 17
}

Value Document

{
  "crepes": {
    "quantity": 598,
    "size": "large"
  }
}