Remove usage of CustomKeyGenerator

Author: codope

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

@@ -54,7 +54,7 @@ public static KeyGenerator createKeyGenerator(TypedProperties props) throws IOEx
     return Objects.isNull(keyGenerator) ? createAvroKeyGeneratorByType(props) : keyGenerator;
   }
 
-  private static KeyGenerator createAvroKeyGeneratorByType(TypedProperties props) throws IOException {
+  public static KeyGenerator createAvroKeyGeneratorByType(TypedProperties props) throws IOException {
     // Use KeyGeneratorType.SIMPLE as default keyGeneratorType
     String keyGeneratorType =
         props.getString(HoodieWriteConfig.KEYGENERATOR_TYPE.key(), null);

hudi-kafka-connect/src/main/java/org/apache/hudi/connect/utils/KafkaConnectUtils.java

@@ -18,8 +18,6 @@
 
 package org.apache.hudi.connect.utils;
 
-import com.google.protobuf.ByteString;
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hudi.client.WriteStatus;
 import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.common.model.HoodieCommitMetadata;
@@ -36,9 +34,11 @@
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.keygen.BaseKeyGenerator;
 import org.apache.hudi.keygen.CustomAvroKeyGenerator;
-import org.apache.hudi.keygen.CustomKeyGenerator;
 import org.apache.hudi.keygen.KeyGenerator;
 import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
+
+import com.google.protobuf.ByteString;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.DescribeTopicsResult;
 import org.apache.kafka.clients.admin.TopicDescription;
@@ -185,8 +185,7 @@ public static String getRecordKeyColumns(KeyGenerator keyGenerator) {
    * @return partition columns Returns the partition columns separated by comma.
    */
   public static String getPartitionColumns(KeyGenerator keyGenerator, TypedProperties typedProperties) {
-
-    if (keyGenerator instanceof CustomKeyGenerator || keyGenerator instanceof CustomAvroKeyGenerator) {
+    if (keyGenerator instanceof CustomAvroKeyGenerator) {
       return ((BaseKeyGenerator) keyGenerator).getPartitionPathFields().stream().map(
           pathField -> Arrays.stream(pathField.split(CustomAvroKeyGenerator.SPLIT_REGEX))
               .findFirst().orElse("Illegal partition path field format: '$pathField' for ${c.getClass.getSimpleName}"))
@@ -200,7 +199,6 @@ public static String getPartitionColumns(KeyGenerator keyGenerator, TypedPropert
     return typedProperties.getString(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key());
   }
 
-
   /**
    * Get the Metadata from the latest commit file.
    *

hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java

@@ -84,7 +84,7 @@ public KafkaConnectTransactionServices(KafkaConnectConfigs connectConfigs) throw
     context = new HoodieJavaEngineContext(hadoopConf);
 
     try {
-      KeyGenerator keyGenerator = HoodieAvroKeyGeneratorFactory.createKeyGenerator(
+      KeyGenerator keyGenerator = HoodieAvroKeyGeneratorFactory.createAvroKeyGeneratorByType(
           new TypedProperties(connectConfigs.getProps()));
       String recordKeyFields = KafkaConnectUtils.getRecordKeyColumns(keyGenerator);
       String partitionColumns = KafkaConnectUtils.getPartitionColumns(keyGenerator,