[HUDI-2863] Rename option 'hoodie.parquet.page.size' to 'write.parquet.page.size' (#4128)

hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java

@@ -417,7 +417,7 @@ public class FlinkOptions extends HoodieConfig {
           + "For DFS, this needs to be aligned with the underlying filesystem block size for optimal performance.");
 
   public static final ConfigOption<Integer> WRITE_PARQUET_PAGE_SIZE = ConfigOptions
-      .key("hoodie.parquet.page.size")
+      .key("write.parquet.page.size")
       .intType()
       .defaultValue(1)
       .withDescription("Parquet page size. Page is the unit of read within a parquet file. "

hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java

@@ -28,7 +28,6 @@ import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.common.engine.EngineType;
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.model.HoodieCleaningPolicy;
-import org.apache.hudi.common.table.HoodieTableConfig;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.log.HoodieLogFormat;
 import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
@@ -233,8 +232,6 @@ public class StreamerUtil {
     Properties properties = new Properties();
     // put all the set options
     flatConf.addAllToProperties(properties);
-    // ugly: table keygen clazz, needed by TwoToThreeUpgradeHandler
-    properties.put(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key(), conf.getString(FlinkOptions.KEYGEN_CLASS_NAME));
     // put all the default options
     for (ConfigOption<?> option : FlinkOptions.optionalOptions()) {
       if (!flatConf.contains(option) && option.hasDefaultValue()) {