fix(QTDI-679): schema collision with Cyrillic names

component-api/src/main/java/org/talend/sdk/component/api/record/Schema.java

@@ -17,20 +17,14 @@
 
 import java.io.StringReader;
 import java.math.BigDecimal;
-import java.nio.charset.Charset;
-import java.nio.charset.CharsetEncoder;
-import java.nio.charset.StandardCharsets;
 import java.time.temporal.Temporal;
 import java.util.Arrays;
-import java.util.Base64;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
@@ -494,53 +488,9 @@ default Schema build(Comparator<Entry> order) {
      *
      * @return avro compatible name.
      */
+    @Deprecated
     static String sanitizeConnectionName(final String name) {
-        if (SKIP_SANITIZE || name == null || name.isEmpty()) {
-            return name;
-        }
-
-        char current = name.charAt(0);
-        final CharsetEncoder ascii = Charset.forName(StandardCharsets.US_ASCII.name()).newEncoder();
-        final boolean skipFirstChar = ((!ascii.canEncode(current)) || (!Character.isLetter(current) && current != '_'))
-                && name.length() > 1 && (!Character.isDigit(name.charAt(1)));
-
-        final StringBuilder sanitizedBuilder = new StringBuilder();
-
-        if (!skipFirstChar) {
-            if (((!Character.isLetter(current)) && current != '_') || (!ascii.canEncode(current))) {
-                sanitizedBuilder.append('_');
-            } else {
-                sanitizedBuilder.append(current);
-            }
-        }
-        for (int i = 1; i < name.length(); i++) {
-            current = name.charAt(i);
-            if (!ascii.canEncode(current)) {
-                if (Character.isLowerCase(current) || Character.isUpperCase(current)) {
-                    sanitizedBuilder.append('_');
-                } else {
-                    final byte[] encoded =
-                            Base64.getEncoder().encode(name.substring(i, i + 1).getBytes(StandardCharsets.UTF_8));
-                    final String enc = new String(encoded);
-                    if (sanitizedBuilder.length() == 0 && Character.isDigit(enc.charAt(0))) {
-                        sanitizedBuilder.append('_');
-                    }
-                    for (int iter = 0; iter < enc.length(); iter++) {
-                        if (Character.isLetterOrDigit(enc.charAt(iter))) {
-                            sanitizedBuilder.append(enc.charAt(iter));
-                        } else {
-                            sanitizedBuilder.append('_');
-                        }
-                    }
-                }
-            } else if (Character.isLetterOrDigit(current)) {
-                sanitizedBuilder.append(current);
-            } else {
-                sanitizedBuilder.append('_');
-            }
-
-        }
-        return sanitizedBuilder.toString();
+        return SchemaCompanionUtil.sanitizeName(name);
     }
 
     @RequiredArgsConstructor
@@ -679,39 +629,13 @@ public int compare(final Entry e1, final Entry e2) {
         }
     }
 
+    /**
+     * Use instead {@since SchemaCompanionUtil#avoidCollision(Schema.Entry, Function, BiConsumer)}
+     */
+    @Deprecated
     static Schema.Entry avoidCollision(final Schema.Entry newEntry,
             final Function<String, Entry> entryGetter,
             final BiConsumer<String, Entry> replaceFunction) {
-        if (SKIP_SANITIZE) {
-            return newEntry;
-        }
-        final Optional<Entry> collisionedEntry = Optional.ofNullable(entryGetter //
-                .apply(newEntry.getName())) //
-                .filter((final Entry field) -> !Objects.equals(field, newEntry));
-        if (!collisionedEntry.isPresent()) {
-            // No collision, return new entry.
-            return newEntry;
-        }
-        final Entry matchedEntry = collisionedEntry.get();
-        final boolean matchedToChange = matchedEntry.getRawName() != null && !(matchedEntry.getRawName().isEmpty());
-        if (matchedToChange) {
-            // the rename has to be applied on entry already inside schema, so replace.
-            replaceFunction.accept(matchedEntry.getName(), newEntry);
-        } else if (newEntry.getRawName() == null || newEntry.getRawName().isEmpty()) {
-            // try to add exactly same raw, skip the add here.
-            return null;
-        }
-        final Entry fieldToChange = matchedToChange ? matchedEntry : newEntry;
-        int indexForAnticollision = 1;
-        final String baseName = Schema.sanitizeConnectionName(fieldToChange.getRawName()); // recalc primiti name.
-
-        String newName = baseName + "_" + indexForAnticollision;
-        while (entryGetter.apply(newName) != null) {
-            indexForAnticollision++;
-            newName = baseName + "_" + indexForAnticollision;
-        }
-        final Entry newFieldToAdd = fieldToChange.toBuilder().withName(newName).build();
-
-        return newFieldToAdd; // matchedToChange ? newFieldToAdd : newEntry;
+        return SchemaCompanionUtil.avoidCollision(newEntry, entryGetter, replaceFunction);
     }
 }

component-api/src/main/java/org/talend/sdk/component/api/record/SchemaCompanionUtil.java

@@ -0,0 +1,165 @@
+/**
+ * Copyright (C) 2006-2025 Talend Inc. - www.talend.com
+ *
+ * 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 org.talend.sdk.component.api.record;
+
+import java.nio.charset.CharsetEncoder;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import org.talend.sdk.component.api.record.Schema.Entry;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class SchemaCompanionUtil {
+
+    /**
+     * Sanitize name to be avro compatible.
+     *
+     * @param name : original name.
+     *
+     * @return avro compatible name.
+     */
+    public static String sanitizeName(final String name) {
+        if (Schema.SKIP_SANITIZE || name == null || name.isEmpty()) {
+            return name;
+        }
+
+        final CharsetEncoder ascii = StandardCharsets.US_ASCII.newEncoder();
+        final StringBuilder sanitizedBuilder = new StringBuilder();
+        final char firstLetter = sanitizeFirstLetter(name, ascii);
+        if (firstLetter != (char) -1) {
+            sanitizedBuilder.append(firstLetter);
+        }
+
+        for (int i = 1; i < name.length(); i++) {
+            char current = name.charAt(i);
+            if (ascii.canEncode(current)) {
+                sanitizedBuilder.append(Character.isLetterOrDigit(current) ? current : '_');
+            } else {
+                if (Character.isLowerCase(current) || Character.isUpperCase(current)) {
+                    sanitizedBuilder.append('_');
+                } else {
+                    final byte[] encoded = base64(name.substring(i, i + 1));
+                    final String enc = new String(encoded, StandardCharsets.UTF_8);
+                    if (sanitizedBuilder.length() == 0 && Character.isDigit(enc.charAt(0))) {
+                        sanitizedBuilder.append('_');
+                    }
+
+                    for (int iter = 0; iter < enc.length(); iter++) {
+                        final char encodedCurrentChar = enc.charAt(iter);
+                        final char sanitizedLetter = Character.isLetterOrDigit(encodedCurrentChar)
+                                ? encodedCurrentChar
+                                : '_';
+                        sanitizedBuilder.append(sanitizedLetter);
+                    }
+                }
+            }
+
+        }
+        return sanitizedBuilder.toString();
+    }
+
+    private static byte[] base64(final String value) {
+        return Base64.getEncoder().encode(value.getBytes(StandardCharsets.UTF_8));
+    }
+
+    private static char sanitizeFirstLetter(final String name, final CharsetEncoder ascii) {
+        char current = name.charAt(0);
+        final boolean skipFirstChar = !(ascii.canEncode(current) && validFirstLetter(current))
+                && name.length() > 1 && !Character.isDigit(name.charAt(1));
+
+        // indicates that first letter is not valid, so it has to be skipped.
+        // and because the next letter is valid (or can be sanitized) we can use it as first letter.
+        if (skipFirstChar) {
+            return (char) -1;
+        }
+
+        if (validFirstLetter(current) && ascii.canEncode(current)) {
+            return current;
+        } else {
+            return '_';
+        }
+    }
+
+    private static boolean validFirstLetter(final char value) {
+        return Character.isLetter(value) || value == '_';
+    }
+
+    /**
+     * May return a different entry with different name.
+     */
+    public static Schema.Entry avoidCollision(final Schema.Entry newEntry,
+            final Function<String, Entry> entryGetter,
+            final BiConsumer<String, Entry> replaceFunction) {
+        if (Schema.SKIP_SANITIZE) {
+            return newEntry;
+        }
+
+        final Entry alreadyExistedEntry = findCollidedEntry(newEntry, entryGetter);
+        if (alreadyExistedEntry == null) {
+            // No collision, return new entry.
+            return newEntry;
+        }
+
+        final boolean matchedToChange = !isEmpty(alreadyExistedEntry.getRawName());
+        if (matchedToChange) {
+            // the rename has to be applied on entry already inside schema, so replace. (dunno why)
+            // replace existed entry with a new name
+            final String newSanitizedName = newNotCollidedName(entryGetter, alreadyExistedEntry.getRawName());
+            final Entry updatedExistedEntry = alreadyExistedEntry.toBuilder()
+                    .withName(newSanitizedName)
+                    .build();
+            replaceFunction.accept(alreadyExistedEntry.getName(), updatedExistedEntry);
+            return newEntry;
+        } else if (isEmpty(newEntry.getRawName())) {
+            // try to add exactly same raw, skip the add here.
+            return null;
+        } else {
+            // raw name isn't empty, so we need to create a new entry with a new name (sanitized).
+            final String newSanitizedName = newNotCollidedName(entryGetter, newEntry.getRawName());
+            return newEntry.toBuilder()
+                    .withName(newSanitizedName)
+                    .build();
+        }
+    }
+
+    private static Entry findCollidedEntry(final Entry newEntry, final Function<String, Entry> entryGetter) {
+        return Optional.ofNullable(entryGetter.apply(newEntry.getName()))
+                .filter(retrievedEntry -> !Objects.equals(retrievedEntry, newEntry))
+                .orElse(null);
+    }
+
+    private static String newNotCollidedName(final Function<String, Entry> entryGetter, final String rawName) {
+        final String baseName = sanitizeName(rawName);
+        int indexForAnticollision = 1;
+        String newName = baseName + "_" + indexForAnticollision;
+        while (entryGetter.apply(newName) != null) {
+            indexForAnticollision++;
+            newName = baseName + "_" + indexForAnticollision;
+        }
+        return newName;
+    }
+
+    private static boolean isEmpty(final String value) {
+        return value == null || value.isEmpty();
+    }
+}