Revert "[HUDI-5575] Adding/Fixing auto generation of record keys w/ hudi (#7726)"

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

@@ -2259,10 +2259,6 @@ public WriteConcurrencyMode getWriteConcurrencyMode() {
     return WriteConcurrencyMode.fromValue(getString(WRITE_CONCURRENCY_MODE));
   }
 
-  public Boolean doAutoGenerateRecordKeys() {
-    return getBooleanOrDefault(KeyGeneratorOptions.AUTO_GENERATE_RECORD_KEYS);
-  }
-
   public boolean isEarlyConflictDetectionEnable() {
     return getBoolean(EARLY_CONFLICT_DETECTION_ENABLE);
   }
@@ -2809,11 +2805,6 @@ public Builder doSkipDefaultPartitionValidation(boolean skipDefaultPartitionVali
       return this;
     }
 
-    public Builder withAutoGenerateRecordKeys(boolean autoGenerateRecordKeys) {
-      writeConfig.setValue(KeyGeneratorOptions.AUTO_GENERATE_RECORD_KEYS, String.valueOf(autoGenerateRecordKeys));
-      return this;
-    }
-
     public Builder withEarlyConflictDetectionEnable(boolean enable) {
       writeConfig.setValue(EARLY_CONFLICT_DETECTION_ENABLE, String.valueOf(enable));
       return this;

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/ComplexAvroKeyGenerator.java

@@ -17,17 +17,13 @@
 
 package org.apache.hudi.keygen;
 
+import org.apache.avro.generic.GenericRecord;
 import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
 
-import org.apache.avro.generic.GenericRecord;
-
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.stream.Collectors;
 
-import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
-
 /**
  * Avro complex key generator, which takes names of fields to be used for recordKey and partitionPath as configs.
  */
@@ -36,26 +32,19 @@ public class ComplexAvroKeyGenerator extends BaseKeyGenerator {
 
   public ComplexAvroKeyGenerator(TypedProperties props) {
     super(props);
-    this.recordKeyFields = props.containsKey(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key())
-        ? Arrays.stream(props.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key()).split(","))
+    this.recordKeyFields = Arrays.stream(props.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key()).split(","))
         .map(String::trim)
         .filter(s -> !s.isEmpty())
-        .collect(Collectors.toList()) : Collections.EMPTY_LIST;
+        .collect(Collectors.toList());
     this.partitionPathFields = Arrays.stream(props.getString(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key()).split(","))
         .map(String::trim)
         .filter(s -> !s.isEmpty())
         .collect(Collectors.toList());
-    instantiateAutoRecordKeyGenerator();
   }
 
   @Override
   public String getRecordKey(GenericRecord record) {
-    if (autoGenerateRecordKeys) {
-      return autoRecordKeyGenerator.getRecordKey(record);
-    } else {
-      checkArgument(getRecordKeyFieldNames().size() > 0, "Record key fields cannot be empty");
-      return KeyGenUtils.getRecordKey(record, getRecordKeyFieldNames(), isConsistentLogicalTimestampEnabled());
-    }
+    return KeyGenUtils.getRecordKey(record, getRecordKeyFieldNames(), isConsistentLogicalTimestampEnabled());
   }
 
   @Override

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/CustomAvroKeyGenerator.java

@@ -43,6 +43,9 @@
  */
 public class CustomAvroKeyGenerator extends BaseKeyGenerator {
 
+  private static final String DEFAULT_PARTITION_PATH_SEPARATOR = "/";
+  public static final String SPLIT_REGEX = ":";
+
   /**
    * Used as a part of config in CustomKeyGenerator.java.
    */
@@ -54,7 +57,6 @@ public CustomAvroKeyGenerator(TypedProperties props) {
     super(props);
     this.recordKeyFields = Arrays.stream(props.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key()).split(",")).map(String::trim).collect(Collectors.toList());
     this.partitionPathFields = Arrays.stream(props.getString(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key()).split(",")).map(String::trim).collect(Collectors.toList());
-    instantiateAutoRecordKeyGenerator();
   }
 
   @Override
@@ -100,14 +102,10 @@ public String getPartitionPath(GenericRecord record) {
 
   @Override
   public String getRecordKey(GenericRecord record) {
-    if (autoGenerateRecordKeys) {
-      return autoRecordKeyGenerator.getRecordKey(record);
-    } else {
-      validateRecordKeyFields();
-      return getRecordKeyFieldNames().size() == 1
-          ? new SimpleAvroKeyGenerator(config).getRecordKey(record)
-          : new ComplexAvroKeyGenerator(config).getRecordKey(record);
-    }
+    validateRecordKeyFields();
+    return getRecordKeyFieldNames().size() == 1
+        ? new SimpleAvroKeyGenerator(config).getRecordKey(record)
+        : new ComplexAvroKeyGenerator(config).getRecordKey(record);
   }
 
   private void validateRecordKeyFields() {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/GlobalAvroDeleteKeyGenerator.java

@@ -36,17 +36,11 @@ public class GlobalAvroDeleteKeyGenerator extends BaseKeyGenerator {
   public GlobalAvroDeleteKeyGenerator(TypedProperties config) {
     super(config);
     this.recordKeyFields = Arrays.asList(config.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key()).split(","));
-    this.partitionPathFields = new ArrayList<>();
-    instantiateAutoRecordKeyGenerator();
   }
 
   @Override
   public String getRecordKey(GenericRecord record) {
-    if (autoGenerateRecordKeys) {
-      return autoRecordKeyGenerator.getRecordKey(record);
-    } else {
-      return KeyGenUtils.getRecordKey(record, getRecordKeyFieldNames(), isConsistentLogicalTimestampEnabled());
-    }
+    return KeyGenUtils.getRecordKey(record, getRecordKeyFieldNames(), isConsistentLogicalTimestampEnabled());
   }
 
   @Override
@@ -56,7 +50,7 @@ public String getPartitionPath(GenericRecord record) {
 
   @Override
   public List<String> getPartitionPathFields() {
-    return partitionPathFields;
+    return new ArrayList<>();
   }
 
   public String getEmptyPartition() {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/KeyGenUtils.java

@@ -108,12 +108,10 @@ public static String getRecordKey(GenericRecord record, List<String> recordKeyFi
         keyIsNullEmpty = false;
       }
     }
-    if (recordKey.length() > 0) {
-      recordKey.deleteCharAt(recordKey.length() - 1);
-    }
+    recordKey.deleteCharAt(recordKey.length() - 1);
     if (keyIsNullEmpty) {
       throw new HoodieKeyException("recordKey values: \"" + recordKey + "\" for fields: "
-          + recordKeyFields + " cannot be entirely null or empty.");
+          + recordKeyFields.toString() + " cannot be entirely null or empty.");
     }
     return recordKey.toString();
   }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/NonpartitionedAvroKeyGenerator.java

@@ -39,7 +39,6 @@ public NonpartitionedAvroKeyGenerator(TypedProperties props) {
     this.recordKeyFields = Arrays.stream(props.getString(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key())
         .split(",")).map(String::trim).filter(s -> !s.isEmpty()).collect(Collectors.toList());
     this.partitionPathFields = EMPTY_PARTITION_FIELD_LIST;
-    instantiateAutoRecordKeyGenerator();
   }
 
   @Override
@@ -54,17 +53,13 @@ public List<String> getPartitionPathFields() {
 
   @Override
   public String getRecordKey(GenericRecord record) {
-    if (autoGenerateRecordKeys) {
-      return autoRecordKeyGenerator.getRecordKey(record);
-    } else {
-      // for backward compatibility, we need to use the right format according to the number of record key fields
-      // 1. if there is only one record key field, the format of record key is just "<value>"
-      // 2. if there are multiple record key fields, the format is "<field1>:<value1>,<field2>:<value2>,..."
-      if (getRecordKeyFieldNames().size() == 1) {
-        return KeyGenUtils.getRecordKey(record, getRecordKeyFieldNames().get(0), isConsistentLogicalTimestampEnabled());
-      }
-      return KeyGenUtils.getRecordKey(record, getRecordKeyFieldNames(), isConsistentLogicalTimestampEnabled());
+    // for backward compatibility, we need to use the right format according to the number of record key fields
+    // 1. if there is only one record key field, the format of record key is just "<value>"
+    // 2. if there are multiple record key fields, the format is "<field1>:<value1>,<field2>:<value2>,..."
+    if (getRecordKeyFieldNames().size() == 1) {
+      return KeyGenUtils.getRecordKey(record, getRecordKeyFieldNames().get(0), isConsistentLogicalTimestampEnabled());
     }
+    return KeyGenUtils.getRecordKey(record, getRecordKeyFieldNames(), isConsistentLogicalTimestampEnabled());
   }
 
   public String getEmptyPartition() {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/keygen/SimpleAvroKeyGenerator.java

@@ -17,15 +17,12 @@
 
 package org.apache.hudi.keygen;
 
+import org.apache.avro.generic.GenericRecord;
 import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
 
-import org.apache.avro.generic.GenericRecord;
-
 import java.util.Collections;
 
-import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
-
 /**
  * Avro simple key generator, which takes names of fields to be used for recordKey and partitionPath as configs.
  */
@@ -46,17 +43,11 @@ public SimpleAvroKeyGenerator(TypedProperties props) {
         ? Collections.emptyList()
         : Collections.singletonList(recordKeyField);
     this.partitionPathFields = Collections.singletonList(partitionPathField);
-    instantiateAutoRecordKeyGenerator();
   }
 
   @Override
   public String getRecordKey(GenericRecord record) {
-    if (autoGenerateRecordKeys) {
-      return autoRecordKeyGenerator.getRecordKey(record);
-    } else {
-      checkArgument(getRecordKeyFieldNames().size() == 1, "Only 1 record key field allowed for SimpleKeyGenerator");
-      return KeyGenUtils.getRecordKey(record, getRecordKeyFieldNames().get(0), isConsistentLogicalTimestampEnabled());
-    }
+    return KeyGenUtils.getRecordKey(record, getRecordKeyFieldNames().get(0), isConsistentLogicalTimestampEnabled());
   }
 
   @Override