public static void setupTimestampKeygenOptions(Configuration conf, DataType fieldType) {
        if (conf.contains(FlinkOptions.KEYGEN_CLASS_NAME)) {
            // the keygen clazz has been set up explicitly, skipping
            return;
        }

        conf.setString(FlinkOptions.KEYGEN_CLASS_NAME, TimestampBasedAvroKeyGenerator.class.getName());
        LOG.info("Table option [{}] is reset to {} because datetime partitioning turns on",
                FlinkOptions.KEYGEN_CLASS_NAME.key(), TimestampBasedAvroKeyGenerator.class.getName());
        if (DataTypeUtils.isTimestampType(fieldType)) {
            int precision = DataTypeUtils.precision(fieldType.getLogicalType());
            if (precision == 0) {
                // seconds
                conf.setString(KeyGeneratorOptions.Config.TIMESTAMP_TYPE_FIELD_PROP,
                        TimestampBasedAvroKeyGenerator.TimestampType.UNIX_TIMESTAMP.name());
            } else if (precision == 3) {
                // milliseconds
                conf.setString(KeyGeneratorOptions.Config.TIMESTAMP_TYPE_FIELD_PROP,
                        TimestampBasedAvroKeyGenerator.TimestampType.EPOCHMILLISECONDS.name());
            }
            String outputPartitionFormat =
                    conf.getOptional(FlinkOptions.PARTITION_FORMAT).orElse(FlinkOptions.PARTITION_FORMAT_HOUR);
            conf.setString(KeyGeneratorOptions.Config.TIMESTAMP_OUTPUT_DATE_FORMAT_PROP, outputPartitionFormat);
        } else {
            conf.setString(KeyGeneratorOptions.Config.TIMESTAMP_TYPE_FIELD_PROP,
                    TimestampBasedAvroKeyGenerator.TimestampType.SCALAR.name());
            conf.setString(KeyGeneratorOptions.Config.INPUT_TIME_UNIT, TimeUnit.DAYS.toString());

            String outputPartitionFormat =
                    conf.getOptional(FlinkOptions.PARTITION_FORMAT).orElse(FlinkOptions.PARTITION_FORMAT_DAY);
            conf.setString(KeyGeneratorOptions.Config.TIMESTAMP_OUTPUT_DATE_FORMAT_PROP, outputPartitionFormat);
            // the option is actually useless, it only works for validation
            conf.setString(KeyGeneratorOptions.Config.TIMESTAMP_INPUT_DATE_FORMAT_PROP,
                    FlinkOptions.PARTITION_FORMAT_DAY);
        }
        conf.setString(KeyGeneratorOptions.Config.TIMESTAMP_OUTPUT_TIMEZONE_FORMAT_PROP, "UTC");
    }
Sets up the keygen options when the partition path is datetime type.

<p>The UTC timezone is used as default.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/hudi/src/main/java/org/apache/inlong/sort/hudi/table/HoodieTableFactory.java