Made the provided UUID code work for keys and values

KAFKA-101

Author: rozza

CHANGELOG.md

@@ -9,6 +9,7 @@
   - [KAFKA-110](https://jira.mongodb.org/browse/KAFKA-110) Added `document.id.strategy.overwrite.existing` configuration.
     Note: This defaults to false, which is a change of behaviour.
   - [KAFKA-118](https://jira.mongodb.org/browse/KAFKA-118) Made UuidStrategy configurable so can output BsonBinary Uuid values
+  - [KAFKA-101](https://jira.mongodb.org/browse/KAFKA-101) Added `UuidProvidedInKeyStrategy` & `UuidProvidedInValueStrategy`
 
 ## 1.1.0
   - [KAFKA-45](https://jira.mongodb.org/browse/KAFKA-45) Allow the Sink connector to ignore unused source record key or value fields.

src/main/java/com/mongodb/kafka/connect/sink/processor/id/strategy/UuidInValueStrategy.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Original Work: Apache License, Version 2.0, Copyright 2017 Hans-Peter Grahsl.
+ */
+package com.mongodb.kafka.connect.sink.processor.id.strategy;
+
+public class UuidProvidedInKeyStrategy extends UuidProvidedStrategy  {
+
+    UuidProvidedInKeyStrategy() {
+        super(ProvidedIn.KEY);
+    }
+
+}

src/main/java/com/mongodb/kafka/connect/sink/processor/id/strategy/UuidProvidedInKeyStrategy.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Original Work: Apache License, Version 2.0, Copyright 2017 Hans-Peter Grahsl.
+ */
+package com.mongodb.kafka.connect.sink.processor.id.strategy;
+
+public class UuidProvidedInValueStrategy extends UuidProvidedStrategy  {
+
+    UuidProvidedInValueStrategy() {
+        super(ProvidedIn.VALUE);
+    }
+
+}

src/main/java/com/mongodb/kafka/connect/sink/processor/id/strategy/UuidProvidedInValueStrategy.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package com.mongodb.kafka.connect.sink.processor.id.strategy;
+
+import static java.lang.String.format;
+
+import java.util.UUID;
+
+import org.apache.kafka.connect.errors.DataException;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+import org.bson.BsonBinary;
+import org.bson.BsonBinarySubType;
+import org.bson.BsonValue;
+import org.bson.UuidRepresentation;
+
+import com.mongodb.kafka.connect.sink.converter.SinkDocument;
+
+class UuidProvidedStrategy extends ProvidedStrategy {
+
+    private static final int UUID_LENGTH = 36;
+    private static final int UUID_LENGTH_NO_DASHES = 32;
+
+    UuidProvidedStrategy(final ProvidedIn where) {
+        super(where);
+    }
+
+    @Override
+    public BsonValue generateId(final SinkDocument doc, final SinkRecord orig) {
+        BsonValue id = super.generateId(doc, orig);
+
+        if (id.isBinary() && BsonBinarySubType.isUuid(id.asBinary().getType())) {
+            return id;
+        } else if (id.isString()) {
+            return new BsonBinary(constructUuidObjectFromString(id.asString().getValue()), UuidRepresentation.STANDARD);
+        }
+
+        throw new DataException(format("UUID cannot be constructed from provided value: `%s`", id));
+    }
+
+    private UUID constructUuidObjectFromString(final String uuid) {
+        try {
+            if (uuid.length() == UUID_LENGTH) {
+                return UUID.fromString(uuid);
+            } else if (uuid.length() == UUID_LENGTH_NO_DASHES) {
+                return UUID.fromString(uuid.replaceFirst(
+                        "(\\p{XDigit}{8})(\\p{XDigit}{4})(\\p{XDigit}{4})(\\p{XDigit}{4})(\\p{XDigit}+)",
+                        "$1-$2-$3-$4-$5"
+                ));
+            }
+        } catch (Exception e) {
+            // ignore
+        }
+
+        throw new DataException(format("UUID cannot be constructed from provided value: `%s`", uuid));
+    }
+}

src/main/java/com/mongodb/kafka/connect/sink/processor/id/strategy/UuidProvidedStrategy.java

@@ -170,15 +170,39 @@ List<DynamicTest> testIdGenerationStrategies() {
             assertEquals(new BsonDocument(), idS6.generateId(sdWithoutKeyDoc, null));
         }));
 
-        IdStrategy idS7 = new UuidInValueStrategy();
-        idTests.add(dynamicTest(UuidInValueStrategy.class.getSimpleName() + " in value", () -> {
+        IdStrategy idS7 = new UuidProvidedInKeyStrategy();
+        idTests.add(dynamicTest(UuidProvidedInKeyStrategy.class.getSimpleName() + " in key", () -> {
+            String idValue = "6d01622d-b3d5-466d-ae48-e414901af8f2";
+            UUID idUuid = UUID.fromString(idValue);
+            SinkDocument sdWithIdInKeyDoc = new SinkDocument(new BsonDocument("_id", new BsonString(idValue)), null);
+            SinkDocument sdWithoutIdInKeyDoc = new SinkDocument(new BsonDocument(), null);
+            SinkDocument sdWithBsonNullIdInKeyDoc = new SinkDocument(new BsonDocument(), null);
+            SinkDocument sdWithInvalidUuidInKeyDoc = new SinkDocument(new BsonDocument("_id", new BsonString("invalid")), null);
+            BsonValue id = idS7.generateId(sdWithIdInKeyDoc, null);
+
+            assertAll("id checks",
+                    () -> assertTrue(id instanceof BsonBinary),
+                    () -> {
+                        BsonBinary bin = (BsonBinary) id;
+                        UUID foundUuid = bin.asUuid(UuidRepresentation.STANDARD);
+                        assertEquals(idUuid, foundUuid);
+                        assertEquals(idValue, foundUuid.toString());
+                    }
+            );
+            assertThrows(DataException.class, () -> idS7.generateId(sdWithoutIdInKeyDoc, null));
+            assertThrows(DataException.class, () -> idS7.generateId(sdWithBsonNullIdInKeyDoc, null));
+            assertThrows(DataException.class, () -> idS7.generateId(sdWithInvalidUuidInKeyDoc, null));
+        }));
+
+        IdStrategy idS8 = new UuidProvidedInValueStrategy();
+        idTests.add(dynamicTest(UuidProvidedInValueStrategy.class.getSimpleName() + " in value", () -> {
             String idValue = "6d01622d-b3d5-466d-ae48-e414901af8f2";
             UUID idUuid = UUID.fromString(idValue);
             SinkDocument sdWithIdInValueDoc = new SinkDocument(null, new BsonDocument("_id", new BsonString(idValue)));
             SinkDocument sdWithoutIdInValueDoc = new SinkDocument(null, new BsonDocument());
             SinkDocument sdWithBsonNullIdInValueDoc = new SinkDocument(null, new BsonDocument());
             SinkDocument sdWithInvalidUuidInValueDoc = new SinkDocument(null, new BsonDocument("_id", new BsonString("invalid")));
-            BsonValue id = idS7.generateId(sdWithIdInValueDoc, null);
+            BsonValue id = idS8.generateId(sdWithIdInValueDoc, null);
 
             assertAll("id checks",
                     () -> assertTrue(id instanceof BsonBinary),
@@ -189,9 +213,9 @@ List<DynamicTest> testIdGenerationStrategies() {
                         assertEquals(idValue, foundUuid.toString());
                     }
             );
-            assertThrows(DataException.class, () -> idS7.generateId(sdWithoutIdInValueDoc, null));
-            assertThrows(DataException.class, () -> idS7.generateId(sdWithBsonNullIdInValueDoc, null));
-            assertThrows(DataException.class, () -> idS7.generateId(sdWithInvalidUuidInValueDoc, null));
+            assertThrows(DataException.class, () -> idS8.generateId(sdWithoutIdInValueDoc, null));
+            assertThrows(DataException.class, () -> idS8.generateId(sdWithBsonNullIdInValueDoc, null));
+            assertThrows(DataException.class, () -> idS8.generateId(sdWithInvalidUuidInValueDoc, null));
         }));
 
         return idTests;