Removes any invalid character from the Avro schema (#11)

Co-authored-by: Emilio Larrambebere <[email protected]>

Author: TPRobots

source/src/main/java/com/trustpilot/connector/dynamodb/utils/RecordConverter.java

@@ -5,6 +5,7 @@
 import com.amazonaws.services.dynamodbv2.model.TableDescription;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Strings;
 import com.trustpilot.connector.dynamodb.Envelope;
 import com.trustpilot.connector.dynamodb.SourceInfo;
 import org.apache.kafka.connect.data.Schema;
@@ -129,6 +130,12 @@ private String sanitiseAttributeName(KeySchemaElement element) {
     }
 
     private String sanitiseAttributeName(final String attributeName) {
-        return attributeName.replaceAll("^[^a-zA-Z_]|(?<!^)[^a-zA-Z0-9_]", "_");
+        final String sanitisedAttributeName = attributeName.replaceAll("^[^a-zA-Z_]|(?<!^)[^a-zA-Z0-9_]", "");
+
+        if (Strings.isNullOrEmpty(sanitisedAttributeName)) {
+            throw new IllegalStateException(String.format("The field name %s couldn't be sanitised correctly", attributeName));
+        }
+
+        return sanitisedAttributeName;
     }
 }

source/src/test/java/com/trustpilot/connector/dynamodb/utils/RecordConverterTests.java

@@ -222,9 +222,9 @@ public void singleItemKeyIsAddedToRecordWhenKeyContainsInvalidCharacters() throw
         );
 
         // Assert
-        assertEquals("test_1234", record.keySchema().fields().get(0).name());
+        assertEquals("test1234", record.keySchema().fields().get(0).name());
         assertEquals(SchemaBuilder.string().build(), record.keySchema().fields().get(0).schema());
-        assertEquals("testKV1Value", ((Struct) record.key()).getString("test_1234"));
+        assertEquals("testKV1Value", ((Struct) record.key()).getString("test1234"));
     }
 
     @Test
@@ -247,13 +247,13 @@ public void multiItemKeyIsAddedToRecordWhenKeyContainsInvalidCharacters() throws
         );
 
         // Assert
-        assertEquals("test_1234", record.keySchema().fields().get(0).name());
+        assertEquals("test1234", record.keySchema().fields().get(0).name());
         assertEquals(SchemaBuilder.string().build(), record.keySchema().fields().get(0).schema());
-        assertEquals("testKV1Value", ((Struct) record.key()).getString("test_1234"));
+        assertEquals("testKV1Value", ((Struct) record.key()).getString("test1234"));
 
-        assertEquals("__starts_with_number", record.keySchema().fields().get(1).name());
+        assertEquals("startswithnumber", record.keySchema().fields().get(1).name());
         assertEquals(SchemaBuilder.string().build(), record.keySchema().fields().get(1).schema());
-        assertEquals("2", ((Struct) record.key()).getString("__starts_with_number"));
+        assertEquals("2", ((Struct) record.key()).getString("startswithnumber"));
     }
 
     @Test
@@ -271,7 +271,7 @@ public void recordAttributesAreAddedToValueDataWhenAttributesContainsInvalidChar
                 "testSequenceNumberID1"
         );
 
-        String expected = "{\"test_1234\":{\"s\":\"testKV1Value\"},\"_starts_with_underscore\":{\"n\":\"1\"},\"__starts_with_number\":{\"s\":\"2\"},\"test______\":{\"s\":\"testStringValue\"}}";
+        String expected = "{\"test1234\":{\"s\":\"testKV1Value\"},\"_starts_with_underscore\":{\"n\":\"1\"},\"startswithnumber\":{\"s\":\"2\"},\"test\":{\"s\":\"testStringValue\"}}";
 
         // Assert
         assertEquals(expected,