Skip to content

Commit d2b0398

Browse files
hpgrahslhpgrahsl
authored
customization for write model filters (#27)
* add feature to be able to configure filter document for replace one model introduced a write model filter strategy to be able to support different use cases. besides the default option there is now a new filter strategy which allows to work with unique business keys and still have e.g. a mongodb ObjectId inserted once during the first upsert per document. this strategy only leads to a correct behaviour in case the business key fields (expressed by means of the PartialValueStrategy for the _id field) are guaranteed to be unique in the original kafka records. see config option mongodb.replace.one.strategy with two pre-defined strategy options - at.grahsl.kafka.connect.mongodb.writemodel.filter.strategy.ReplaceOneDefaultFilterStrategy - at.grahsl.kafka.connect.mongodb.writemodel.filter.strategy.ReplaceOneBusinessKeyFilterStrategy * document new feature allowing to configure write model filters * add basic unit tests for write model strategies * fix delete one model default strategy
1 parent cf9ffb8 commit d2b0398

File tree

9 files changed

+418
-35
lines changed

9 files changed

+418
-35
lines changed

‎README.md‎

Lines changed: 86 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -355,6 +355,74 @@ These settings cause:
355355

356356
Note the use of the **"." character** as navigational operator in both examples. It's used in order to refer to nested fields in sub documents of the record structure. The prefix at the very beginning is used as a simple convention to distinguish between the _key_ and _value_ structure of a document.
357357

358+
### Custom Write Model Filters
359+
The default behaviour for the connector whenever documents are written to MongoDB collections is to make use of a proper [ReplaceOneModel](http://mongodb.github.io/mongo-java-driver/3.6/javadoc/com/mongodb/client/model/ReplaceOneModel.html) with [upsert mode](http://mongodb.github.io/mongo-java-driver/3.6/javadoc/com/mongodb/client/model/UpdateOptions.html) and **create the filter documents based on the _id field** as it is given in the value structure of the sink record.
360+
However, there are other use cases which need a different approach and the **customization option for generating filter documents** can support these.
361+
A new configuration option (_mongodb.replace.one.strategy_) allows for such customizations. Currently, the following two strategies are implemented:
362+
363+
* **default behaviour** at.grahsl.kafka.connect.mongodb.writemodel.filter.strategy.**ReplaceOneDefaultFilterStrategy**
364+
* **business key** (see description of use case below) at.grahsl.kafka.connect.mongodb.writemodel.filter.strategy.**ReplaceOneBusinessKeyFilterStrategy**
365+
366+
##### Use Case: Employing Business Keys
367+
Let's say you want to re-use a unique business key found in your sink records while at the same time have _BSON ObjectIds_ created for the resulting MongoDB documents.
368+
To achieve this a few simple configuration steps are necessary:
369+
370+
1) make sure to **create a unique key constraint** for the business key of your target MongoDB collection
371+
2) use the **PartialValueStrategy** as the DocumentIdAdder's strategy in order to let the connector know which fields belong to the business key
372+
3) use the **ReplaceOneBusinessKeyFilterStrategy** instead of the default behaviour
373+
374+
These configuration settings then allow to have **filter documents based on the original business key but still have _BSON ObjectIds_ created for the _id field** during the first upsert into your target MongoDB target collection.
375+
Find below how such a setup might look like:
376+
377+
Given the following fictional Kafka record
378+
379+
```json
380+
{
381+
"fieldA": "Anonymous",
382+
"fieldB": 42,
383+
"active": true,
384+
"values": [12.34, 23.45, 34.56, 45.67]
385+
}
386+
```
387+
388+
together with the sink connector config below
389+
390+
```json
391+
{
392+
"name": "mdb-sink",
393+
"config": {
394+
"key.converter":"org.apache.kafka.connect.json.JsonConverter",
395+
"key.converter.schemas.enable": false,
396+
"value.converter":"org.apache.kafka.connect.json.JsonConverter",
397+
"value.converter.schemas.enable": false,
398+
"connector.class": "at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector",
399+
"topics": "mytopic",
400+
"mongodb.connection.uri": "mongodb://mongodb:27017/kafkaconnect?w=1&journal=true",
401+
"mongodb.document.id.strategy": "at.grahsl.kafka.connect.mongodb.processor.id.strategy.PartialValueStrategy",
402+
"mongodb.key.projection.list": "fieldA,fieldB",
403+
"mongodb.key.projection.type": "whitelist",
404+
"mongodb.collection": "mycollection",
405+
"mongodb.replace.one.strategy":"at.grahsl.kafka.connect.mongodb.writemodel.filter.strategy.ReplaceOneBusinessKeyFilterStrategy"
406+
}
407+
}
408+
```
409+
410+
will eventually result in MongoDB documents looking like:
411+
412+
```json
413+
{
414+
"_id": ObjectId("5abf52cc97e51aae0679d237"),
415+
"fieldA": "Anonymous",
416+
"fieldB": 42,
417+
"active": true,
418+
"values": [12.34, 23.45, 34.56, 45.67]
419+
}
420+
```
421+
422+
All upsert operations are done based on the unique business key which for this example is a compound one that consists of the two fields _(fieldA,fieldB)_.
423+
424+
NOTE: Future versions will allow to make use of arbitrary, individual strategies that can be registered and as used for the _mongodb.replace.one.strategy_ configuration setting.
425+
358426
### Change Data Capture Mode
359427
The sink connector can also be used in a different operation mode in order to handle change data capture (CDC) events. Currently, the following CDC events from [Debezium](http://debezium.io/) can be processed:
360428

@@ -404,23 +472,24 @@ Data is written using acknowledged writes and the configured write concern level
404472

405473
At the moment the following settings can be configured by means of the *connector.properties* file. For a config file containing default settings see [this example](https://github.com/hpgrahsl/kafka-connect-mongodb/blob/master/config/MongoDbSinkConnector.properties).
406474

407-
| Name | Description | Type | Default | Valid Values | Importance |
408-
|-------------------------------------|----------------------------------------------------------------------------------------|---------|-----------------------------------------------------------------------|------------------------------|------------|
409-
| mongodb.collection | single sink collection name to write to | string | kafkatopic | | high |
410-
| mongodb.connection.uri | the mongodb connection URI as supported by the official drivers | string | mongodb://localhost:27017/kafkaconnect?w=1&journal=true | | high |
411-
| mongodb.document.id.strategy | class name of strategy to use for generating a unique document id (_id) | string | at.grahsl.kafka.connect.mongodb.processor.id.strategy.BsonOidStrategy | | high |
412-
| mongodb.delete.on.null.values | whether or not the connector tries to delete documents based on key when value is null | boolean | false | | medium |
413-
| mongodb.max.num.retries | how often a retry should be done on write errors | int | 3 | [0,...] | medium |
414-
| mongodb.retries.defer.timeout | how long in ms a retry should get deferred | int | 5000 | [0,...] | medium |
415-
| mongodb.change.data.capture.handler | class name of CDC handler to use for processing | string | "" | | low |
416-
| mongodb.document.id.strategies | comma separated list of custom strategy classes to register for usage | string | "" | | low |
417-
| mongodb.field.renamer.mapping | inline JSON array with objects describing field name mappings (see docs) | string | [] | | low |
418-
| mongodb.field.renamer.regexp | inline JSON array with objects describing regexp settings (see docs) | string | [] | | low |
419-
| mongodb.key.projection.list | comma separated list of field names for key projection | string | "" | | low |
420-
| mongodb.key.projection.type | whether or not and which key projection to use | string | none | [none, blacklist, whitelist] | low |
421-
| mongodb.post.processor.chain | comma separated list of post processor classes to build the chain with | string | at.grahsl.kafka.connect.mongodb.processor.DocumentIdAdder | | low |
422-
| mongodb.value.projection.list | comma separated list of field names for value projection | string | "" | | low |
423-
| mongodb.value.projection.type | whether or not and which value projection to use | string | none | [none, blacklist, whitelist] | low |
475+
| Name | Description | Type | Default | Valid Values | Importance |
476+
|-------------------------------------|----------------------------------------------------------------------------------------|---------|--------------------------------------------------------------------------------------------|------------------------------|------------|
477+
| mongodb.collection | single sink collection name to write to | string | kafkatopic | | high |
478+
| mongodb.connection.uri | the monogdb connection URI as supported by the offical drivers | string | mongodb://localhost:27017/kafkaconnect?w=1&journal=true | | high |
479+
| mongodb.document.id.strategy | class name of strategy to use for generating a unique document id (_id) | string | at.grahsl.kafka.connect.mongodb.processor.id.strategy.BsonOidStrategy | | high |
480+
| mongodb.delete.on.null.values | whether or not the connector tries to delete documents based on key when value is null | boolean | false | | medium |
481+
| mongodb.max.num.retries | how often a retry should be done on write errors | int | 3 | [0,...] | medium |
482+
| mongodb.retries.defer.timeout | how long in ms a retry should get deferred | int | 5000 | [0,...] | medium |
483+
| mongodb.change.data.capture.handler | class name of CDC handler to use for processing | string | "" | | low |
484+
| mongodb.document.id.strategies | comma separated list of custom strategy classes to register for usage | string | "" | | low |
485+
| mongodb.field.renamer.mapping | inline JSON array with objects describing field name mappings (see docs) | string | [] | | low |
486+
| mongodb.field.renamer.regexp | inline JSON array with objects describing regexp settings (see docs) | string | [] | | low |
487+
| mongodb.key.projection.list | comma separated list of field names for key projection | string | "" | | low |
488+
| mongodb.key.projection.type | whether or not and which key projection to use | string | none | [none, blacklist, whitelist] | low |
489+
| mongodb.post.processor.chain | comma separated list of post processor classes to build the chain with | string | at.grahsl.kafka.connect.mongodb.processor.DocumentIdAdder | | low |
490+
| mongodb.replace.one.strategy | how to build the filter doc for the replaceOne write model | string | at.grahsl.kafka.connect.mongodb.writemodel.filter.strategy.ReplaceOneDefaultFilterStrategy | | low |
491+
| mongodb.value.projection.list | comma separated list of field names for value projection | string | "" | | low |
492+
| mongodb.value.projection.type | whether or not and which value projection to use | string | none | [none, blacklist, whitelist] | low |
424493

425494
### Running in development
426495

‎config/MongoDbSinkConnector.properties‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,3 +42,4 @@ mongodb.field.renamer.regexp=[]
4242
mongodb.post.processor.chain=at.grahsl.kafka.connect.mongodb.processor.DocumentIdAdder
4343
mongodb.change.data.capture.handler=
4444
mongodb.delete.on.null.values=false
45+
mongodb.replace.one.strategy=at.grahsl.kafka.connect.mongodb.writemodel.filter.strategy.ReplaceOneDefaultFilterStrategy

‎src/main/java/at/grahsl/kafka/connect/mongodb/MongoDbSinkConnectorConfig.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import at.grahsl.kafka.connect.mongodb.processor.field.renaming.RegExpSettings;
2727
import at.grahsl.kafka.connect.mongodb.processor.field.renaming.RenameByRegExp;
2828
import at.grahsl.kafka.connect.mongodb.processor.id.strategy.*;
29+
import at.grahsl.kafka.connect.mongodb.writemodel.filter.strategy.WriteModelFilterStrategy;
2930
import com.fasterxml.jackson.core.type.TypeReference;
3031
import com.fasterxml.jackson.databind.ObjectMapper;
3132
import com.mongodb.MongoClientURI;
@@ -64,6 +65,7 @@ public enum FieldProjectionTypes {
6465
public static final String MONGODB_POST_PROCESSOR_CHAIN_DEFAULT = "at.grahsl.kafka.connect.mongodb.processor.DocumentIdAdder";
6566
public static final String MONGODB_CHANGE_DATA_CAPTURE_HANDLER_DEFAULT = "";
6667
public static final boolean MONGODB_DELETE_ON_NULL_VALUES_DEFAULT = false;
68+
public static final String MONGODB_REPLACE_ONE_STRATEGY_DEFAULT = "at.grahsl.kafka.connect.mongodb.writemodel.filter.strategy.ReplaceOneDefaultFilterStrategy";
6769

6870
public static final String MONGODB_CONNECTION_URI_CONF = "mongodb.connection.uri";
6971
private static final String MONGODB_CONNECTION_URI_DOC = "the monogdb connection URI as supported by the offical drivers";
@@ -110,6 +112,9 @@ public enum FieldProjectionTypes {
110112
public static final String MONGODB_DELETE_ON_NULL_VALUES = "mongodb.delete.on.null.values";
111113
private static final String MONGODB_DELETE_ON_NULL_VALUES_DOC = "whether or not the connector tries to delete documents based on key when value is null";
112114

115+
public static final String MONGODB_REPLACE_ONE_STRATEGY = "mongodb.replace.one.strategy";
116+
private static final String MONGODB_REPLACE_ONE_STRATEGY_DOC = "how to build the filter doc for the replaceOne write model";
117+
113118
private static ObjectMapper objectMapper = new ObjectMapper();
114119

115120
public MongoDbSinkConnectorConfig(ConfigDef config, Map<String, String> parsedConfig) {
@@ -137,6 +142,7 @@ public static ConfigDef conf() {
137142
.define(MONGODB_POST_PROCESSOR_CHAIN, Type.STRING, MONGODB_POST_PROCESSOR_CHAIN_DEFAULT, Importance.LOW, MONGODB_POST_PROCESSOR_CHAIN_DOC)
138143
.define(MONGODB_CHANGE_DATA_CAPTURE_HANDLER, Type.STRING, MONGODB_CHANGE_DATA_CAPTURE_HANDLER_DEFAULT, Importance.LOW, MONGODB_CHANGE_DATA_CAPTURE_HANDLER_DOC)
139144
.define(MONGODB_DELETE_ON_NULL_VALUES, Type.BOOLEAN, MONGODB_DELETE_ON_NULL_VALUES_DEFAULT, Importance.MEDIUM, MONGODB_DELETE_ON_NULL_VALUES_DOC)
145+
.define(MONGODB_REPLACE_ONE_STRATEGY, Type.STRING, MONGODB_REPLACE_ONE_STRATEGY_DEFAULT, Importance.LOW, MONGODB_REPLACE_ONE_STRATEGY_DOC)
140146
;
141147
}
142148

@@ -310,6 +316,20 @@ public boolean isDeleteOnNullValues() {
310316
return getBoolean(MONGODB_DELETE_ON_NULL_VALUES);
311317
}
312318

319+
public WriteModelFilterStrategy getReplaceOneFilterStrategy() {
320+
String replaceOneFilterStrategy = getString(MONGODB_REPLACE_ONE_STRATEGY);
321+
try {
322+
return (WriteModelFilterStrategy) Class.forName(replaceOneFilterStrategy)
323+
.getConstructor().newInstance();
324+
} catch (ReflectiveOperationException e) {
325+
throw new ConfigException(e.getMessage(),e);
326+
} catch (ClassCastException e) {
327+
throw new ConfigException("error: specified class "+ replaceOneFilterStrategy
328+
+ " violates the contract since it doesn't implement " +
329+
WriteModelFilterStrategy.class);
330+
}
331+
}
332+
313333
public static Set<String> getPredefinedCdcHandlerClassNames() {
314334
Set<String> cdcHandlers = new HashSet<String>();
315335
cdcHandlers.add(MongoDbHandler.class.getName());

0 commit comments

Comments
 (0)