Add regular expressions in topic mapping (#141)

Author: stIncMale

config/checkstyle/suppressions.xml

@@ -40,4 +40,5 @@
     <suppress checks="[a-zA-Z0-9]*" files="com[\\/]mongodb[\\/]kafka[\\/]connect[\\/]avro[\\/]" />
     <suppress checks="[a-zA-Z0-9]*" files="com[\\/]mongodb[\\/]kafka[\\/]connect[\\/]embedded[\\/]" />
     <suppress checks="[a-zA-Z0-9]*" files="com[\\/]mongodb[\\/]kafka[\\/]connect[\\/]mongodb[\\/]" />
+    <suppress checks="MethodLength" files=".*DefaultTopicMapperTest.java"/>
 </suppressions>

src/main/java/com/mongodb/kafka/connect/source/MongoSourceConfig.java

@@ -184,15 +184,76 @@ public class MongoSourceConfig extends AbstractConfig {
   public static final String TOPIC_NAMESPACE_MAP_CONFIG = "topic.namespace.map";
   private static final String TOPIC_NAMESPACE_MAP_DISPLAY = "The namespace to topic map";
   private static final String TOPIC_NAMESPACE_MAP_DOC =
-      "A json map that maps change stream document namespaces to topics.\n"
-          + "For example: `{\"db\": \"dbTopic\", \"db.coll\": \"dbCollTopic\"}` will map all "
-          + "change stream documents from the `db` database to `dbTopic.<collectionName>` apart from"
-          + "any documents from the `db.coll` namespace which map to the `dbCollTopic` topic.\n"
-          + "If you want to map all messages to a single topic use `*`: "
-          + "For example: `{\"*\": \"everyThingTopic\", \"db.coll\": \"exceptionToTheRuleTopic\"}` "
-          + "will map all change stream documents to the `everyThingTopic` apart from the `db.coll` "
-          + "messages."
-          + "Note: Any prefix and suffix configuration will still apply.";
+      "A JSON object specifying how to map a MongoDB change stream document namespace"
+          + " to a Kafka topic name. Used by the `DefaultTopicMapper`."
+          + " MongoDB change stream document namespace is"
+          + " a database name optionally concatenated with a collection name, separated by full stop '.'."
+          + "\nThe name in each JSON name/value pair is a namespace pattern,"
+          + " the value is a string representing the corresponding topic name template."
+          + " Pairs are ordered. When there are multiple pairs with equal namespace patterns (duplicates),"
+          + " they are deduplicated and only one pair is taken into account:"
+          + " its position is taken from the first pair among duplicates,"
+          + " its topic name template is taken from the last pair among duplicates."
+          + " After deduplication, pairs with an empty topic name template are ignored when computing topic names."
+          + " Note that a topic name computed based on this configuration is then decorated"
+          + " using the `topic.prefix` and `topic.suffix` configuration properties."
+          + "\nThere are three kinds of pairs:"
+          + "\n- Simple. The namespace pattern must not contain solidus '/' and can be either of the following:"
+          + "\n  - A namespace with a collection name, in which case it matches only that namespace."
+          + " The topic name template is interpreted as the topic name."
+          + "\n  - A namespace with only a database name, in which case it matches any namespace having that database name."
+          + " The matching namespace may either have a collection name, or not:"
+          + "\n    - If there is a collection name, then the topic name is computed"
+          + " by concatenating the topic name template and the collection name from the matching namespace, separated by `topic.separator`."
+          + "\n    - If there is no collection name, then the topic name template is interpreted as the topic name."
+          + "\n- Regex. The namespace pattern starts with solidus '/',"
+          + " followed by a regular expression with the syntax and behavior as per `java.util.regex.Pattern`."
+          + " The topic name is computed by doing variable expansion on the topic name template."
+          + " The following variables are supported:"
+          + "\n  - `db` The database name from the matching namespace."
+          + "\n  - `sep` The value of the `topic.separator` configuration property."
+          + "\n  - `coll` The collection name from the matching namespace, or an empty string if there is no collection name."
+          + "\n  - `sep_coll` The value of the `coll` variable"
+          + " prefixed with the value of `sep` if and only if the value of `coll` is not empty."
+          + "\n  - `coll_sep` The value of the `coll` variable"
+          + " suffixed with the value of `sep` if and only if the value of `coll` is not empty."
+          + "\n  - `sep_coll_sep` The value of the `coll` variable"
+          + " prefixed and suffixed with the value of `sep` if and only if the value of `coll` is not empty."
+          + "\n To be expanded, a variable must be enclosed between curly brackets '{' and '}', for example '{db}'."
+          + " The characters '{' and '}' are not allowed to be used in the topic name template for any other purpose."
+          + " Note that variable names are case-sensitive."
+          + "\nBe careful when creating a namespace pattern with characters that need escaping according to the JSON syntax."
+          + " For example, if you want to match full stop '.', the regex syntax requires escaping it as '\\.'."
+          + " However, reverse solidus '\\' itself must be escaped as '\\\\' according to the JSON syntax."
+          + " Consequently, to match '.' you need to write '\\\\.'."
+          + "\n- Wildcard. The namespace pattern is asterisk '*' and matches any namespace."
+          + " The topic name template is interpreted as the topic name."
+          + "\n The matching order:"
+          + "\n1. Simple pairs with a collection name in the namespace pattern."
+          + "\n2. Simple pairs without a collection name in the namespace pattern."
+          + "\n3. Regex pairs in order."
+          + "\n4. The wildcard pair."
+          + "\n Matching stops as soon as the first match is found. If no matches are found,"
+          + " the topic name is computed solely based on the namespace. The namespace may either have a collection name, or not:"
+          + "\n- If there is a collection name, then the topic name is computed"
+          + " by concatenating the database name and the collection name, separated by `topic.separator`."
+          + "\n- If there is no collection name, then the database name is used as the topic name."
+          + "\nExamples (`topic.separator` is assumed to be '-'):"
+          + "\n1. '{\"myDb\": \"topicTwo\", \"myDb.myColl\": \"topicOne\"}'"
+          + " The 'myDb.myColl' namespace is mapped to the 'topicOne' topic name."
+          + " All other namespaces with the 'myDb' database name are mapped"
+          + " either to the 'topicTwo-<collection name>' topic name, if they have a collection name,"
+          + " or to the 'topicTwo' topic otherwise."
+          + " All other namespaces are mapped"
+          + " either to the '<database name>-<collection name>' topic name, if they have a collection name,"
+          + " or to the '<database name>' topic otherwise."
+          + "\n2. '{\"/myDb(?:\\\\..*)?\": \"topicTwo{sep_coll}\", \"*\": \"topicThree\", \"myDb.myColl\": \"topicOne\"}'"
+          + " The regex namespace pattern matches any namespace with the 'myDb' database name."
+          + " The 'myDb.myColl' namespace is mapped to the 'topicOne' topic name."
+          + " All other namespaces with the 'myDb' database name are mapped"
+          + " either to the 'topicTwo-<collection name>' topic name, if they have a collection name,"
+          + " or to the 'topicTwo' topic otherwise."
+          + " All other namespaces are mapped to the 'topicThree' topic name.";
   private static final String TOPIC_NAMESPACE_MAP_DEFAULT = EMPTY_STRING;
 
   public static final String PIPELINE_CONFIG = "pipeline";