Kafka Sink Connector Post-Processors¶

DocumentIdAdder¶

The DocumentIdAdder post processor provides the _id field for a SinkDocument before it is written to a MongoDB collection. This post processor is configured using a strategy that contains the logic for generating the value of the _id. The following strategies are provided with this connector:

You can assign the document.id.strategy property as follows:

document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.UuidStrategy

To define a custom strategy, create a class that implements the interface IdStrategy and provide the fully-qualified path to the document.id.strategy setting.

Block List / Allow List Projector¶

This section provides example projection configurations to show how they filter the following sample record:

{
  "name": "Anonymous",
  "age": 42,
  "active": true,
  "address": {
    "city": "Unknown",
    "country": "NoWhereLand"
  },
  "food": [
    "Austrian",
    "Italian"
  ],
  "data": [
    {
      "k": "foo",
      "v": 1
    }
  ],
  "lut": {
    "key1": 12.34,
    "key2": 23.45
  },
  "destination: {
    "city": "Springfield",
    "country": "AnotherLand"
  }
}

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

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

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

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

{
  "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"
  }
}

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

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

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

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

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

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

[key|value].projection.type=BlockList
[key|value].projection.list=**.high
{
  "city": "Springfield",
  "population: {
    "qty": 30.7,
    "scale": 1000,
    "units": "people"
  }
}

Field Renaming Post Processors¶

This section provides example configurations for the RenameByMapping and RenameByRegex post processors to show how they update field names in a sink record. The field renaming parameters specify whether to update the key or value document in the record using dot notation as well as the pattern to match. You must specify the RenameByMapping and RenameByRegex properties with your parameters in a JSON array.

The field renaming post processor examples use the following sample sink record:

Sample Key Document

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

Sample Value Document

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

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

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

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

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

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

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

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

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

Create Your Own Custom Write Model Strategy¶

If none of the pre-built write strategies suits your needs, you can create your own. A custom write model strategy is a Java class that implements WriteModelStrategy.

To configure your Sink Connector to use the custom WriteModelStrategy, follow the steps below:

1. Create a class that implements the WriteModelStrategy interface and overrides the createWriteModel(SinkDocument) method.
2. Compile the class to a JAR file.

3. Add the compiled JAR to the Class Path / Plugin Path for Kafka workers. For more information about plugin paths, see the Confluent documentation.

   Note

   Plugins are loaded in isolation, so when deploying a custom write strategy, both the connector JAR and the custom write model strategy JAR should be on the same path as shown in the following example:

   mongodb-kafka-connect/lib/mongo-kafka-connect-all.jar

   mongodb-kafka-connect/lib/custom-write-model-strategy.jar
4. Specify your Java class in the writemodel.strategy configuration setting.

The following is an example of a custom write strategy. It extracts the value of fullDocument from the value document and returns a ReplaceOne operation. The ReplaceOne operation replaces a document in the MongoDB collection that matches the ID from the value document.

/**
 * A custom write model strategy
 *
 * This example takes the 'fullDocument' field from a change stream and creates a
 * ReplaceOne operation.
 */
public class CustomWriteModelStrategy implements WriteModelStrategy {
  private static String ID = "_id";
  @Override
  public WriteModel<BsonDocument> createWriteModel(final SinkDocument document) {
    BsonDocument changeStreamDocument = document.getValueDoc()
        .orElseThrow(() -> new DataException("Missing value document"));
    BsonDocument fullDocument = changeStreamDocument.getDocument("fullDocument", new BsonDocument());
    if (fullDocument.isEmpty()) {
      return null; // Return null to indicate no op.
    }
    return new ReplaceOneModel<>(Filters.eq(ID, fullDocument.get(ID)), fullDocument);
  }
}

WriteModel Strategy: Business Keys¶

A business key is a value comprised of data within the sink record that identifies it as a unique document. This example defines a business key using data contained within multiple fields in the record as well as instructs the post processor to generate BSON ObjectIds 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.

For this example, we 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
  }
}

WriteModel Strategy: Inserted and Modified Timestamps¶

This example shows how we can track the create and update timestamps on documents that are upserted by the connector. The UpdateOneTimestampsStrategy custom write model strategy performs the following tasks:

• When a new MongoDB document is inserted into the collection by the connector, the _insertedTS and _modifiedTS fields will be set to the current time.
• When an existing MongoDB document is updated in the collection by the connector, the _modifiedTS field will be updated to the current time.

This is set by specifying the UpdateOneTimestampsStrategy in the configuration file as follows:

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

For this example, we want to track the position of a train along its route. The _modifiedTS field provides us with the time and date that the first position report was saved into the collection.

The position report includes the following data in the value document:

{
  "_id": "MN-1234",
  "start": "Beacon",
  "destination": "Grand Central"
  "position": [ 40.8051693, -73.9388079 ]
}

The writemodel strategy is set to UpdateOneTimestampsStrategy to append the created and modified timestamps and the document id strategy is set to ProvidedInValueStrategy to identify the train using the _id field of the value document.

document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.ProvidedInValueStrategy
writemodel.strategy=com.mongodb.kafka.connect.sink.writemodel.strategy.UpdateOneTimestampsStrategy

The MongoDB document inserted after processing the initial message for the train includes the following data:

{
  "_id": "MN-1234",
  "_insertedTS": ISODate('2019-10-23T15:08:000Z"),
  "_modifiedTS": ISODate("2019-10-23T15:08:000Z"),
  "start": "Beacon",
  "destination": "Grand Central"
  "position": [ 40.805, -73.939 ]
}

After an hour, the train reports its current location along its route. The position and _modifiedTS fields are updated:

{
  "_id": "MN-1234",
  "_insertedTS": ISODate('2019-10-23T15:08:000Z"),
  "_modifiedTS": ISODate("2019-10-23T16:08:000Z"),
  "start": "Beacon",
  "destination": "Grand Central"
  "position": [ 41.156, -73.870 ]
}