KAFKA-425: Add new operation types for change events (#165)

Co-authored-by: Calvin Thomas Nix <[email protected]>

Author: ksshlee

src/main/java/com/mongodb/kafka/connect/sink/cdc/mongodb/OperationType.java

@@ -20,12 +20,19 @@
 
 // https://docs.mongodb.com/manual/reference/change-events/
 public enum OperationType {
+  CREATE_COLLECTION("create"),
+  CREATE_INDEXES("createIndexes"),
   INSERT("insert"),
   REPLACE("replace"),
+  REFINE_COLLECTION_SHARD_KEY("refineCollectionShardKey"),
+  RESHARD_COLLECTION("reshardCollection"),
+  SHARD_COLLECTION("shardCollection"),
   UPDATE("update"),
+  MODIFY_COLLECTION("modify"),
   DELETE("delete"),
   DROP_COLLECTION("drop"),
   DROP_DATABASE("dropDatabase"),
+  DROP_INDEXES("dropIndexes"),
   RENAME_COLLECTION("rename"),
   INVALIDATE("invalidate"),
   UNKNOWN("unknown");

src/test/java/com/mongodb/kafka/connect/sink/cdc/mongodb/OperationTypeTest.java

@@ -102,6 +102,69 @@ Stream<DynamicTest> dynamicTestsFromCollection() {
                   () -> assertEquals(OperationType.INVALIDATE, operationType),
                   () -> assertEquals(value, operationType.getValue()));
             }),
+        dynamicTest(
+            "create",
+            () -> {
+              String value = "create";
+              OperationType operationType = OperationType.fromString(value);
+              assertAll(
+                  () -> assertEquals(OperationType.CREATE_COLLECTION, operationType),
+                  () -> assertEquals(value, operationType.getValue()));
+            }),
+        dynamicTest(
+            "createIndexes",
+            () -> {
+              String value = "createIndexes";
+              OperationType operationType = OperationType.fromString(value);
+              assertAll(
+                  () -> assertEquals(OperationType.CREATE_INDEXES, operationType),
+                  () -> assertEquals(value, operationType.getValue()));
+            }),
+        dynamicTest(
+            "refineCollectionShardKey",
+            () -> {
+              String value = "refineCollectionShardKey";
+              OperationType operationType = OperationType.fromString(value);
+              assertAll(
+                  () -> assertEquals(OperationType.REFINE_COLLECTION_SHARD_KEY, operationType),
+                  () -> assertEquals(value, operationType.getValue()));
+            }),
+        dynamicTest(
+            "reshardCollection",
+            () -> {
+              String value = "reshardCollection";
+              OperationType operationType = OperationType.fromString(value);
+              assertAll(
+                  () -> assertEquals(OperationType.RESHARD_COLLECTION, operationType),
+                  () -> assertEquals(value, operationType.getValue()));
+            }),
+        dynamicTest(
+            "shardCollection",
+            () -> {
+              String value = "shardCollection";
+              OperationType operationType = OperationType.fromString(value);
+              assertAll(
+                  () -> assertEquals(OperationType.SHARD_COLLECTION, operationType),
+                  () -> assertEquals(value, operationType.getValue()));
+            }),
+        dynamicTest(
+            "modify",
+            () -> {
+              String value = "modify";
+              OperationType operationType = OperationType.fromString(value);
+              assertAll(
+                  () -> assertEquals(OperationType.MODIFY_COLLECTION, operationType),
+                  () -> assertEquals(value, operationType.getValue()));
+            }),
+        dynamicTest(
+            "dropIndexes",
+            () -> {
+              String value = "dropIndexes";
+              OperationType operationType = OperationType.fromString(value);
+              assertAll(
+                  () -> assertEquals(OperationType.DROP_INDEXES, operationType),
+                  () -> assertEquals(value, operationType.getValue()));
+            }),
         dynamicTest(
             "unknown",
             () -> {