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

This reverts commit deadfd6.

Author: emilio-larrambebere-TP

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

@@ -5,7 +5,6 @@
 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;
@@ -130,12 +129,6 @@ private String sanitiseAttributeName(KeySchemaElement element) {
     }
 
     private String sanitiseAttributeName(final String attributeName) {
-        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;
+        return attributeName.replaceAll("^[^a-zA-Z_]|(?<!^)[^a-zA-Z0-9_]", "_");
     }
 }

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

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