File tree Expand file tree Collapse file tree
hudi-flink-datasource/hudi-flink/src
main/java/org/apache/hudi/util
test/java/org/apache/hudi/utils Expand file tree Collapse file tree Original file line number Diff line number Diff line change 5454import org .apache .hudi .configuration .OptionsResolver ;
5555import org .apache .hudi .exception .HoodieException ;
5656import org .apache .hudi .exception .HoodieIOException ;
57+ import org .apache .hudi .keygen .SimpleAvroKeyGenerator ;
5758import org .apache .hudi .schema .FilebasedSchemaProvider ;
5859import org .apache .hudi .sink .transform .ChainedTransformer ;
5960import org .apache .hudi .sink .transform .Transformer ;
@@ -315,7 +316,8 @@ public static HoodieTableMetaClient initTableIfNotExists(
315316 .setPreCombineField (OptionsResolver .getPreCombineField (conf ))
316317 .setArchiveLogFolder (ARCHIVELOG_FOLDER .defaultValue ())
317318 .setPartitionFields (conf .getString (FlinkOptions .PARTITION_PATH_FIELD , null ))
318- .setKeyGeneratorClassProp (conf .getString (FlinkOptions .KEYGEN_CLASS_NAME ))
319+ .setKeyGeneratorClassProp (
320+ conf .getOptional (FlinkOptions .KEYGEN_CLASS_NAME ).orElse (SimpleAvroKeyGenerator .class .getName ()))
319321 .setHiveStylePartitioningEnable (conf .getBoolean (FlinkOptions .HIVE_STYLE_PARTITIONING ))
320322 .setUrlEncodePartitioning (conf .getBoolean (FlinkOptions .URL_ENCODE_PARTITIONING ))
321323 .setTimelineLayoutVersion (1 )
Original file line number Diff line number Diff line change 2323import org .apache .hudi .common .table .view .FileSystemViewStorageType ;
2424import org .apache .hudi .common .util .FileIOUtils ;
2525import org .apache .hudi .configuration .FlinkOptions ;
26+ import org .apache .hudi .keygen .SimpleAvroKeyGenerator ;
2627import org .apache .hudi .util .StreamerUtil ;
2728import org .apache .hudi .util .ViewStorageProperties ;
2829
@@ -67,6 +68,7 @@ void testInitTableIfNotExists() throws IOException {
6768 "Missing partition columns in the hoodie.properties." );
6869 assertArrayEquals (metaClient1 .getTableConfig ().getPartitionFields ().get (), new String [] {"p0" , "p1" });
6970 assertEquals (metaClient1 .getTableConfig ().getPreCombineField (), "ts" );
71+ assertEquals (metaClient1 .getTableConfig ().getKeyGeneratorClassName (), SimpleAvroKeyGenerator .class .getName ());
7072
7173 // Test for non-partitioned table.
7274 conf .removeConfig (FlinkOptions .PARTITION_PATH_FIELD );
@@ -77,6 +79,7 @@ void testInitTableIfNotExists() throws IOException {
7779 .setConf (new org .apache .hadoop .conf .Configuration ())
7880 .build ();
7981 assertFalse (metaClient2 .getTableConfig ().getPartitionFields ().isPresent ());
82+ assertEquals (metaClient2 .getTableConfig ().getKeyGeneratorClassName (), SimpleAvroKeyGenerator .class .getName ());
8083 }
8184
8285 @ Test
You can’t perform that action at this time.
0 commit comments