Kafka Sink Connector Post-Processors¶

Block List Projection Example¶

In the following example sink configuration, we specify a block list projection and the specific fields to omit from the record:

post.processor.chain=com.mongodb.kafka.connect.sink.processor.BlockList[Key|Value]Projector
[key|value].projection.type=BlockList
[key|value].projection.list=age,address.city,lut.key2,data.v

The record contains the following data after applying the projection:

{
  "name": "Anonymous",
  "active": true,
  "address": {
    "country": "NoWhereLand"
  },
  "food": [
    "Austrian",
    "Italian"
  ],
  "data": [
    {
      "k": "foo"
    }
  ],
  "lut": {
    "key1": 12.34
  },
  "destination: {
    "city": "Springfield",
    "country": "AnotherLand"
  }
}

Allow List Projection Example¶

In the following example sink configuration, we specify an allow list projection and the specific fields to include in the record:

post.processor.chain=com.mongodb.kafka.connect.sink.processor.AllowList[Key|Value]Projector
[key|value].projection.type=AllowList
[key|value].projection.list=age,address.city,lut.key2,data.v

The record contains the following data after applying the projection:

{
  "age": 42,
  "address": {
    "city": "Unknown"
  },
  "data": [
    {
      "v": 1
    }
  ],
  "lut": {
    "key2": 23.45
  }
}

Projection Wildcard Pattern Matching¶

The previous example projection configurations demonstrated exact string matching on field names. The projection list setting also supports the following wildcard patterns matching on field names:

• "*" (star): matches a string of any length for the level in the document in which it is specified.
• "**" (double star): matches the current and all nested levels from which it is specified.

The examples below demonstrate how to use each wildcard pattern and the projection output from the following sample record:

{
  "forecast": [
    { "day": "Monday",
      "temp": {
        "degrees": 25.3,
        "units": "C",
        "range": {
          "low": 23.0,
          "high": 28.7
        }
      },
      "uv": 5
    }
  ],
  "city": "Springfield",
  "population: {
    "qty": 30.7,
    "scale": 1000,
    "units": "people"
  }
}

Allow List Wildcard Examples

The * wildcard pattern in the example below matches all the keys named temp in the forecast array and all fields nested a single level below it.

[key|value].projection.type=AllowList
[key|value].projection.list=city,forecast.temp.*

The record contains the following data after applying the projection:

{
  "city": "Springfield",
  "forecast": [
    {
      "temp": {
        "degrees": 25.3,
        "units": "C",
        "range": {
          "low": 23.0,
          "high": 28.7
        }
      }
    }
  ]
}

The ** wildcard pattern in the example below matches all the keys for all levels that contain the field scale.

[key|value].projection.type=AllowList
[key|value].projection.list=**.scale

The record contains the following data after applying the projection:

{
  "population: {
    "qty": 30.7,
    "scale": 1000,
    "units": "people"
  }
}

Block List Wildcard Examples

The wildcard character can also be used to match all field names at specific levels as demonstrated in the following block list projection configuration example:

[key|value].projection.type=BlockList
[key|value].projection.list=population,forecast.*.*

The record contains the following data after applying the projection:

{
  "forecast": [
    {
      "day": "Monday",
      "uv": 5
    }
  ],
  "city": "Springfield",
}

The connector configuration also supports the ** (double star) wildcard which matches the current and all nested levels from which it is specified.

[key|value].projection.type=BlockList
[key|value].projection.list=**.high

The record contains the following data after applying the projection:

{
  "city": "Springfield",
  "population: {
    "qty": 30.7,
    "scale": 1000,
    "units": "people"
  }
}

RenameByMapping Example¶

The RenameByMapping post processor setting specifies one or more objects that assign fields matching a string to a new name in a Key or Value document.

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 transforms the original key document to the following one after applying the RenameByMapping post processor:

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

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

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

This transforms the original value document to the following one after applying the RenameByMapping post processor:

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

You can also specify multiple mappings in the field.renamer.mapping property by using a stringified JSON array as shown in the following example:

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

RenameByRegex¶

The RenameByRegex post processor setting is an array of objects. Each object in the array contains the following JSON element keys:

Example

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

When the post processor is applied to the sample key document and the sample value document, it produces the following output:

Key Document

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

Value Document

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

The post processor applied the following changes:

• All field names in the key document of the sink record that started with "date" are matched. In the matched fields, all instances of "_" are replaced with "-".
• All field names in the value document of the sink record that are subdocuments of crepes are matched. In the matched fields, all instances of "purchased" are replaced with "quantity".

ReplaceOneBusinessKeyStrategy Example¶

In this example, we assemble a business key from multiple fields of a record and instruct the post processor to generate BSON ObjectId instances for inserts, but not for updates.

Follow the steps below to configure this strategy:

1. Create a unique index that corresponds to your business key in your target MongoDB collection.
2. In the connector configuration, specify the PartialValueStrategy as the id strategy to identify the fields that belong to the business key.
3. In the connector configuration, specify the ReplaceOneBusinessKeyStrategy writemodel strategy.

Suppose our task is to track airplane capacity by the flight number and airport location represented by flight_no and airport_code, respectively. An example message contains the following:

{
  "flight_no": "Z342",
  "airport_code": "LAX",
  "passengers": {
    "capacity": 180,
    "occupied": 152
  }
}

To implement the strategy, we first create a unique index on the flight_no and airport_code fields in the MongoDB shell:

db.collection.createIndex({ "flight_no": 1, "airport_code": 1}, { unique: true })

Next, we specify the PartialValueStrategy strategy and fields to include in the business key and specify the ReplaceOneBusinessKeyStrategy writemodel strategy in the configuration file:

document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy
document.id.strategy.partial.value.projection.list=flight_no,airport_code
document.id.strategy.partial.value.projection.type=AllowList
writemodel.strategy=com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy"

The sample data inserted into the collection contains the following:

{
  "_id": ObjectId('5db0b81199b7fbcc89631d06'),
  "flight_no": "Z342"
  "airport_code": "LAX",
  "passengers_spaces": {
    "capacity": 180,
    "occupied": 152
  }
}

When the connector processes sink data that matches an existing document with the same business key field values, it updates the document with the new values without changing the _id field.

{
  "_id": ObjectId('5db0b81199b7fbcc89631d06'),
  "flight_no": "Z342"
  "airport_code": "LAX",
  "passengers_spaces": {
    "capacity": 180,
    "occupied": 95
  }
}

DeleteOneBusinessKeyStrategy Example¶

In this example, we show how to use the DeleteOneBusinessKeyStrategy to do the following:

• Assemble a business key from a single field from a record.
• Instruct the post processor to delete a document in the target collection that contains a value matching the one in the business key.

Suppose our task is to track airline passenger ticket reservations and cancellations. We can use the DeleteOneBusinessKeyStrategy to delete reservations when customers cancel or when reservations go unclaimed.

The following MongoDB document represents a ticket reservation in the tickets database and reservations collection:

{
  "_id": ObjectId('5db0b81199b7fbcc89631d07'),
  "flight_no": "Z920",
  "airport_code": "CDG",
  "reservation_no": "RT211583W"
}

The following steps show how we can configure the value of the reservation_no field as the business key and create the delete strategy:

1. Create a unique index that corresponds to your business key in your target MongoDB collection.
2. In the connector configuration, specify the PartialValueStrategy as the id strategy to identify the field that you want to use as the business key.
3. In the connector configuration, specify the DeleteOneBusinessKeyStrategy writemodel strategy.
4. Specify the source topic and target MongoDB database and collection.

To implement the DeleteOneBusinessKeyStrategy with our connector, first we create a unique index on the reservation_no field in the MongoDB shell:

db.collection.createIndex({ "reservation_no": 1 }, { unique: true })

Next, we specify the PartialValueStrategy strategy as the id strategy and use the projection setting to consider only the reservation_no field with the following configuration:

document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy
document.id.strategy.partial.value.projection.list=reservation_no
document.id.strategy.partial.value.projection.type=AllowList

We specify the strategy in the configuration as follows:

writemodel.strategy=com.mongodb.kafka.connect.sink.writemodel.strategy.DeleteOneBusinessKeyStrategy

We define the Kafka topic from which we receive cancellations and the target ticket reservation database and collection in the configuration file as follows:

topics=ReservationCancellations
database=tickets
collection=reservations

When the connector receives sink data from the "ReservationCancellations" topic, it extracts the business key using the strategy configuration settings and performs a delete operation on the target collection. If the reservation_no field contains a value of "RT211583W", the connector deletes deletes the matching document from the collection.