[HUDI-8554] Use COMMIT_TIME_ORDERING (#12304)

* Use commit time ordering

* address comments

hudi-cli/src/main/java/org/apache/hudi/cli/commands/BootstrapCommand.java

@@ -75,7 +75,7 @@ public String bootstrap(
       @ShellOption(value = {"--payloadClass"}, defaultValue = "",
           help = "Payload Class (deprecated). Use `--merge-mode` to specify the equivalent payload update behavior.") final String payloadClass,
       @ShellOption(value = {"--merge-mode", "--record-merge-mode"}, defaultValue = "",
-          help = "Merge mode to use. 'EVENT_TIME_ORDERING', 'OVERWRITE_WITH_LATEST', "
+          help = "Merge mode to use. 'EVENT_TIME_ORDERING', 'COMMIT_TIME_ORDERING', "
               + "or 'CUSTOM' if you want to set a custom merge strategy ID and implementation.") final String recordMergeMode,
       @ShellOption(value = {"--merge-strategy-id", "--record-merge-strategy-id"}, defaultValue = "",
           help = "ID of the merge strategy to use. Only set when using 'CUSTOM' merge mode") final String recordMergeStrategyId,

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/OverwriteWithLatestSparkRecordMerger.java

@@ -35,7 +35,7 @@ public class OverwriteWithLatestSparkRecordMerger extends HoodieSparkRecordMerge
 
   @Override
   public String getMergingStrategy() {
-    return OVERWRITE_MERGE_STRATEGY_UUID;
+    return COMMIT_TIME_BASED_MERGE_STRATEGY_UUID;
   }
 
   @Override

hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java

@@ -56,11 +56,11 @@ public abstract class BaseSparkInternalRowReaderContext extends HoodieReaderCont
   @Override
   public Option<HoodieRecordMerger> getRecordMerger(RecordMergeMode mergeMode, String mergeStrategyId, String mergeImplClasses) {
     // TODO(HUDI-7843):
-    // get rid of event time and overwrite with latest. Just return Option.empty
+    // get rid of event time and commit time ordering. Just return Option.empty
     switch (mergeMode) {
       case EVENT_TIME_ORDERING:
         return Option.of(new DefaultSparkRecordMerger());
-      case OVERWRITE_WITH_LATEST:
+      case COMMIT_TIME_ORDERING:
         return Option.of(new OverwriteWithLatestSparkRecordMerger());
       case CUSTOM:
       default:

hudi-common/src/main/java/org/apache/hudi/common/config/RecordMergeMode.java

@@ -25,7 +25,7 @@
 public enum RecordMergeMode {
   @EnumFieldDescription("Using transaction time to merge records, i.e., the record from later "
       + "transaction overwrites the earlier record with the same key.")
-  OVERWRITE_WITH_LATEST,
+  COMMIT_TIME_ORDERING,
 
   @EnumFieldDescription("Using event time as the ordering to merge records, i.e., the record "
       + "with the larger event time overwrites the record with the smaller event time on the "

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieRecordMerger.java

@@ -47,7 +47,7 @@ public interface HoodieRecordMerger extends Serializable {
   String DEFAULT_MERGE_STRATEGY_UUID = "eeb8d96f-b1e4-49fd-bbf8-28ac514178e5";
 
   // Always chooses the most recently written record
-  String OVERWRITE_MERGE_STRATEGY_UUID = "ce9acb64-bde0-424c-9b91-f6ebba25356d";
+  String COMMIT_TIME_BASED_MERGE_STRATEGY_UUID = "ce9acb64-bde0-424c-9b91-f6ebba25356d";
 
   // Use avro payload to merge records
   String PAYLOAD_BASED_MERGE_STRATEGY_UUID = "00000000-0000-0000-0000-000000000000";

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieRecordPayload.java

@@ -152,7 +152,7 @@ static String getAvroPayloadForMergeMode(RecordMergeMode mergeMode) {
       default:
       case EVENT_TIME_ORDERING:
         return DefaultHoodieRecordPayload.class.getName();
-      case OVERWRITE_WITH_LATEST:
+      case COMMIT_TIME_ORDERING:
         return OverwriteWithLatestAvroPayload.class.getName();
     }
   }

hudi-common/src/main/java/org/apache/hudi/common/model/OverwriteWithLatestMerger.java

@@ -44,6 +44,6 @@ public HoodieRecord.HoodieRecordType getRecordType() {
 
   @Override
   public String getMergingStrategy() {
-    return OVERWRITE_MERGE_STRATEGY_UUID;
+    return COMMIT_TIME_BASED_MERGE_STRATEGY_UUID;
   }
 }

hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java

@@ -78,7 +78,7 @@
 import static org.apache.hudi.common.config.HoodieReaderConfig.RECORD_MERGE_IMPL_CLASSES_WRITE_CONFIG_KEY;
 import static org.apache.hudi.common.config.RecordMergeMode.CUSTOM;
 import static org.apache.hudi.common.config.RecordMergeMode.EVENT_TIME_ORDERING;
-import static org.apache.hudi.common.config.RecordMergeMode.OVERWRITE_WITH_LATEST;
+import static org.apache.hudi.common.config.RecordMergeMode.COMMIT_TIME_ORDERING;
 import static org.apache.hudi.common.config.TimestampKeyGeneratorConfig.DATE_TIME_PARSER;
 import static org.apache.hudi.common.config.TimestampKeyGeneratorConfig.INPUT_TIME_UNIT;
 import static org.apache.hudi.common.config.TimestampKeyGeneratorConfig.TIMESTAMP_INPUT_DATE_FORMAT;
@@ -89,7 +89,7 @@
 import static org.apache.hudi.common.config.TimestampKeyGeneratorConfig.TIMESTAMP_TIMEZONE_FORMAT;
 import static org.apache.hudi.common.config.TimestampKeyGeneratorConfig.TIMESTAMP_TYPE_FIELD;
 import static org.apache.hudi.common.model.HoodieRecordMerger.DEFAULT_MERGE_STRATEGY_UUID;
-import static org.apache.hudi.common.model.HoodieRecordMerger.OVERWRITE_MERGE_STRATEGY_UUID;
+import static org.apache.hudi.common.model.HoodieRecordMerger.COMMIT_TIME_BASED_MERGE_STRATEGY_UUID;
 import static org.apache.hudi.common.model.HoodieRecordMerger.PAYLOAD_BASED_MERGE_STRATEGY_UUID;
 import static org.apache.hudi.common.util.ConfigUtils.fetchConfigs;
 import static org.apache.hudi.common.util.ConfigUtils.recoverIfNeeded;
@@ -740,8 +740,8 @@ public static Triple<RecordMergeMode, String, String> inferCorrectMergingBehavio
         }
 
         // set merger strategy based on merge mode
-        if (inferredRecordMergeMode == OVERWRITE_WITH_LATEST) {
-          inferredRecordMergeStrategyId = OVERWRITE_MERGE_STRATEGY_UUID;
+        if (inferredRecordMergeMode == COMMIT_TIME_ORDERING) {
+          inferredRecordMergeStrategyId = COMMIT_TIME_BASED_MERGE_STRATEGY_UUID;
         } else if (inferredRecordMergeMode == EVENT_TIME_ORDERING) {
           inferredRecordMergeStrategyId = DEFAULT_MERGE_STRATEGY_UUID;
         } else {
@@ -754,10 +754,10 @@ public static Triple<RecordMergeMode, String, String> inferCorrectMergingBehavio
           checkArgument(recordMergeMode == null || recordMergeMode == EVENT_TIME_ORDERING,
               "Default merge strategy ID can only be used with the merge mode of EVENT_TIME_ORDERING");
           inferredRecordMergeMode = EVENT_TIME_ORDERING;
-        } else if (recordMergeStrategyId.equals(OVERWRITE_MERGE_STRATEGY_UUID)) {
-          checkArgument(recordMergeMode == null || recordMergeMode == OVERWRITE_WITH_LATEST,
-              "Overwrite with latest merger strategy ID can only be used with the merge mode of OVERWRITE_WITH_LATEST");
-          inferredRecordMergeMode = OVERWRITE_WITH_LATEST;
+        } else if (recordMergeStrategyId.equals(COMMIT_TIME_BASED_MERGE_STRATEGY_UUID)) {
+          checkArgument(recordMergeMode == null || recordMergeMode == COMMIT_TIME_ORDERING,
+              "Commit time ordering merger strategy ID can only be used with the merge mode of COMMIT_TIME_ORDERING");
+          inferredRecordMergeMode = COMMIT_TIME_ORDERING;
         } else {
           checkArgument(!recordMergeStrategyId.equals(PAYLOAD_BASED_MERGE_STRATEGY_UUID),
               "Payload based strategy should only be used if you have a custom payload class set");
@@ -786,18 +786,18 @@ public static Triple<RecordMergeMode, String, String> inferCorrectMergingBehavio
           // currently for the custom case. This block will be moved below and check if the payload class name is dummy
           checkArgument(recordMergeMode == null || recordMergeMode == CUSTOM, "Record merge mode, payload class, and merge strategy are in an illegal configuration");
           checkArgument(
-              !recordMergeStrategyId.equals(OVERWRITE_MERGE_STRATEGY_UUID) && !recordMergeStrategyId.equals(PAYLOAD_BASED_MERGE_STRATEGY_UUID),
+              !recordMergeStrategyId.equals(COMMIT_TIME_BASED_MERGE_STRATEGY_UUID) && !recordMergeStrategyId.equals(PAYLOAD_BASED_MERGE_STRATEGY_UUID),
               "Record merger strategy is incompatible with payload class");
           inferredRecordMergeMode = CUSTOM;
           inferredRecordMergeStrategyId = recordMergeStrategyId;
         }
       } else if (payloadClassName.equals(OverwriteWithLatestAvroPayload.class.getName())) {
         // strategy and merge mode must be unset or align with overwrite
-        checkArgument(isNullOrEmpty(recordMergeStrategyId) || recordMergeStrategyId.equals(OVERWRITE_MERGE_STRATEGY_UUID),
+        checkArgument(isNullOrEmpty(recordMergeStrategyId) || recordMergeStrategyId.equals(COMMIT_TIME_BASED_MERGE_STRATEGY_UUID),
             "Record merge strategy cannot be set if a merge payload is used");
-        checkArgument(recordMergeMode == null || recordMergeMode == OVERWRITE_WITH_LATEST, "Only overwrite with latest record merge mode can be used with overwrite payload");
-        inferredRecordMergeMode = OVERWRITE_WITH_LATEST;
-        inferredRecordMergeStrategyId = OVERWRITE_MERGE_STRATEGY_UUID;
+        checkArgument(recordMergeMode == null || recordMergeMode == COMMIT_TIME_ORDERING, "Only commit time ordering merge mode can be used with overwrite payload");
+        inferredRecordMergeMode = COMMIT_TIME_ORDERING;
+        inferredRecordMergeStrategyId = COMMIT_TIME_BASED_MERGE_STRATEGY_UUID;
       } else {
         // using custom avro payload
         checkArgument(isNullOrEmpty(recordMergeStrategyId) || recordMergeStrategyId.equals(PAYLOAD_BASED_MERGE_STRATEGY_UUID),

hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieBaseFileGroupRecordBuffer.java

@@ -112,7 +112,7 @@ public HoodieBaseFileGroupRecordBuffer(HoodieReaderContext<T> readerContext,
       this.payloadClass = Option.empty();
     }
     this.orderingFieldName = Option.ofNullable(ConfigUtils.getOrderingField(props)).orElseGet(() -> hoodieTableMetaClient.getTableConfig().getPreCombineField());
-    this.orderingFieldTypeOpt = recordMergeMode == RecordMergeMode.OVERWRITE_WITH_LATEST ? Option.empty() : AvroSchemaUtils.findNestedFieldType(readerSchema, this.orderingFieldName);
+    this.orderingFieldTypeOpt = recordMergeMode == RecordMergeMode.COMMIT_TIME_ORDERING ? Option.empty() : AvroSchemaUtils.findNestedFieldType(readerSchema, this.orderingFieldName);
     this.orderingFieldDefault = orderingFieldTypeOpt.map(type -> readerContext.castValue(0, type)).orElse(0);
     this.props = props;
     this.internalSchema = readerContext.getSchemaHandler().getInternalSchema();
@@ -218,7 +218,7 @@ protected Option<Pair<T, Map<String, Object>>> doProcessNextDataRecord(T record,
         return Option.empty();
       } else {
         switch (recordMergeMode) {
-          case OVERWRITE_WITH_LATEST:
+          case COMMIT_TIME_ORDERING:
             return Option.empty();
           case EVENT_TIME_ORDERING:
             Comparable existingOrderingValue = readerContext.getOrderingValue(
@@ -295,7 +295,7 @@ protected Option<DeleteRecord> doProcessNextDeletedRecord(DeleteRecord deleteRec
                                                             Pair<Option<T>, Map<String, Object>> existingRecordMetadataPair) {
     if (existingRecordMetadataPair != null) {
       switch (recordMergeMode) {
-        case OVERWRITE_WITH_LATEST:
+        case COMMIT_TIME_ORDERING:
           return Option.empty();
         case EVENT_TIME_ORDERING:
         case CUSTOM:
@@ -400,7 +400,7 @@ protected Option<T> merge(Option<T> older, Map<String, Object> olderInfoMap,
       return Option.empty();
     } else {
       switch (recordMergeMode) {
-        case OVERWRITE_WITH_LATEST:
+        case COMMIT_TIME_ORDERING:
           return newer;
         case EVENT_TIME_ORDERING:
           Comparable oldOrderingValue = readerContext.getOrderingValue(

hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodiePositionBasedFileGroupRecordBuffer.java

@@ -173,7 +173,7 @@ public void processDeleteBlock(HoodieDeleteBlock deleteBlock) throws IOException
     }
 
     switch (recordMergeMode) {
-      case OVERWRITE_WITH_LATEST:
+      case COMMIT_TIME_ORDERING:
         for (Long recordPosition : recordPositions) {
           records.putIfAbsent(recordPosition,
               Pair.of(Option.empty(), readerContext.generateMetadataForRecord(null, "", orderingFieldDefault, orderingFieldTypeOpt)));

hudi-common/src/test/java/org/apache/hudi/common/table/read/TestHoodieFileGroupReaderBase.java

@@ -103,8 +103,8 @@ public void validateRecordsInFileGroup(String tablePath,
 
   private static Stream<Arguments> testArguments() {
     return Stream.of(
-        arguments(RecordMergeMode.OVERWRITE_WITH_LATEST, "avro"),
-        arguments(RecordMergeMode.OVERWRITE_WITH_LATEST, "parquet"),
+        arguments(RecordMergeMode.COMMIT_TIME_ORDERING, "avro"),
+        arguments(RecordMergeMode.COMMIT_TIME_ORDERING, "parquet"),
         arguments(RecordMergeMode.EVENT_TIME_ORDERING, "avro"),
         arguments(RecordMergeMode.EVENT_TIME_ORDERING, "parquet"),
         arguments(RecordMergeMode.CUSTOM, "avro"),

hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java

@@ -56,9 +56,9 @@
 
 import static org.apache.hudi.common.config.RecordMergeMode.CUSTOM;
 import static org.apache.hudi.common.config.RecordMergeMode.EVENT_TIME_ORDERING;
-import static org.apache.hudi.common.config.RecordMergeMode.OVERWRITE_WITH_LATEST;
+import static org.apache.hudi.common.config.RecordMergeMode.COMMIT_TIME_ORDERING;
 import static org.apache.hudi.common.model.HoodieRecordMerger.DEFAULT_MERGE_STRATEGY_UUID;
-import static org.apache.hudi.common.model.HoodieRecordMerger.OVERWRITE_MERGE_STRATEGY_UUID;
+import static org.apache.hudi.common.model.HoodieRecordMerger.COMMIT_TIME_BASED_MERGE_STRATEGY_UUID;
 import static org.apache.hudi.common.model.HoodieRecordMerger.PAYLOAD_BASED_MERGE_STRATEGY_UUID;
 import static org.apache.hudi.common.table.HoodieTableConfig.RECORD_MERGE_MODE;
 import static org.apache.hudi.common.table.HoodieTableConfig.TABLE_CHECKSUM;
@@ -317,16 +317,16 @@ private static Stream<Arguments> argumentsForInferringRecordMergeMode() {
         arguments(null, null, DEFAULT_MERGE_STRATEGY_UUID, false, EVENT_TIME_ORDERING, defaultPayload, DEFAULT_MERGE_STRATEGY_UUID),
 
         //test legal overwrite combos
-        arguments(OVERWRITE_WITH_LATEST, null, null, false, OVERWRITE_WITH_LATEST, overwritePayload, OVERWRITE_MERGE_STRATEGY_UUID),
-        arguments(OVERWRITE_WITH_LATEST, overwritePayload, null, false, OVERWRITE_WITH_LATEST, overwritePayload, OVERWRITE_MERGE_STRATEGY_UUID),
-        arguments(OVERWRITE_WITH_LATEST, overwritePayload, OVERWRITE_MERGE_STRATEGY_UUID, false, OVERWRITE_WITH_LATEST, overwritePayload,
-            OVERWRITE_MERGE_STRATEGY_UUID),
-        arguments(OVERWRITE_WITH_LATEST, null, OVERWRITE_MERGE_STRATEGY_UUID, false, OVERWRITE_WITH_LATEST, overwritePayload,
-            OVERWRITE_MERGE_STRATEGY_UUID),
-        arguments(null, overwritePayload, null, false, OVERWRITE_WITH_LATEST, overwritePayload, OVERWRITE_MERGE_STRATEGY_UUID),
-        arguments(null, overwritePayload, OVERWRITE_MERGE_STRATEGY_UUID, false, OVERWRITE_WITH_LATEST, overwritePayload,
-            OVERWRITE_MERGE_STRATEGY_UUID),
-        arguments(null, null, OVERWRITE_MERGE_STRATEGY_UUID, false, OVERWRITE_WITH_LATEST, overwritePayload, OVERWRITE_MERGE_STRATEGY_UUID),
+        arguments(COMMIT_TIME_ORDERING, null, null, false, COMMIT_TIME_ORDERING, overwritePayload, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID),
+        arguments(COMMIT_TIME_ORDERING, overwritePayload, null, false, COMMIT_TIME_ORDERING, overwritePayload, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID),
+        arguments(COMMIT_TIME_ORDERING, overwritePayload, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, false, COMMIT_TIME_ORDERING, overwritePayload,
+            COMMIT_TIME_BASED_MERGE_STRATEGY_UUID),
+        arguments(COMMIT_TIME_ORDERING, null, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, false, COMMIT_TIME_ORDERING, overwritePayload,
+            COMMIT_TIME_BASED_MERGE_STRATEGY_UUID),
+        arguments(null, overwritePayload, null, false, COMMIT_TIME_ORDERING, overwritePayload, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID),
+        arguments(null, overwritePayload, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, false, COMMIT_TIME_ORDERING, overwritePayload,
+            COMMIT_TIME_BASED_MERGE_STRATEGY_UUID),
+        arguments(null, null, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, false, COMMIT_TIME_ORDERING, overwritePayload, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID),
 
         //test legal custom payload combos
         arguments(CUSTOM, customPayload, null, false, CUSTOM, customPayload, PAYLOAD_BASED_MERGE_STRATEGY_UUID),
@@ -346,25 +346,25 @@ private static Stream<Arguments> argumentsForInferringRecordMergeMode() {
         //test illegal combos
         arguments(EVENT_TIME_ORDERING, overwritePayload, null, true, null, null, null),
         arguments(EVENT_TIME_ORDERING, customPayload, null, true, null, null, null),
-        arguments(EVENT_TIME_ORDERING, null, OVERWRITE_MERGE_STRATEGY_UUID, true, null, null, null),
+        arguments(EVENT_TIME_ORDERING, null, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, true, null, null, null),
         arguments(EVENT_TIME_ORDERING, null, customStrategy, true, null, null, null),
         arguments(EVENT_TIME_ORDERING, null, PAYLOAD_BASED_MERGE_STRATEGY_UUID, true, null, null, null),
-        arguments(OVERWRITE_WITH_LATEST, defaultPayload, null, true, null, null, null),
-        arguments(OVERWRITE_WITH_LATEST, customPayload, null, true, null, null, null),
-        arguments(OVERWRITE_WITH_LATEST, null, DEFAULT_MERGE_STRATEGY_UUID, true, null, null, null),
-        arguments(OVERWRITE_WITH_LATEST, null, customStrategy, true, null, null, null),
-        arguments(OVERWRITE_WITH_LATEST, null, PAYLOAD_BASED_MERGE_STRATEGY_UUID, true, null, null, null),
+        arguments(COMMIT_TIME_ORDERING, defaultPayload, null, true, null, null, null),
+        arguments(COMMIT_TIME_ORDERING, customPayload, null, true, null, null, null),
+        arguments(COMMIT_TIME_ORDERING, null, DEFAULT_MERGE_STRATEGY_UUID, true, null, null, null),
+        arguments(COMMIT_TIME_ORDERING, null, customStrategy, true, null, null, null),
+        arguments(COMMIT_TIME_ORDERING, null, PAYLOAD_BASED_MERGE_STRATEGY_UUID, true, null, null, null),
         arguments(CUSTOM, defaultPayload, null, true, null, null, null),
         arguments(CUSTOM, overwritePayload, null, true, null, null, null),
         arguments(CUSTOM, null, DEFAULT_MERGE_STRATEGY_UUID, true, null, null, null),
-        arguments(CUSTOM, null, OVERWRITE_MERGE_STRATEGY_UUID, true, null, null, null),
+        arguments(CUSTOM, null, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, true, null, null, null),
         arguments(CUSTOM, defaultPayload, PAYLOAD_BASED_MERGE_STRATEGY_UUID, true, null, null, null),
         arguments(CUSTOM, overwritePayload, PAYLOAD_BASED_MERGE_STRATEGY_UUID, true, null, null, null),
-        arguments(CUSTOM, defaultPayload, OVERWRITE_MERGE_STRATEGY_UUID, true, null, null, null),
+        arguments(CUSTOM, defaultPayload, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, true, null, null, null),
         arguments(CUSTOM, overwritePayload, DEFAULT_MERGE_STRATEGY_UUID, true, null, null, null),
         arguments(null, defaultPayload, PAYLOAD_BASED_MERGE_STRATEGY_UUID, true, null, null, null),
         arguments(null, overwritePayload, PAYLOAD_BASED_MERGE_STRATEGY_UUID, true, null, null, null),
-        arguments(null, defaultPayload, OVERWRITE_MERGE_STRATEGY_UUID, true, null, null, null),
+        arguments(null, defaultPayload, COMMIT_TIME_BASED_MERGE_STRATEGY_UUID, true, null, null, null),
         arguments(null, overwritePayload, DEFAULT_MERGE_STRATEGY_UUID, true, null, null, null));
     return arguments;
   }

hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/read/TestOverwriteWithLatestMerger.java

@@ -59,8 +59,8 @@ public class TestOverwriteWithLatestMerger extends HoodieFileGroupReaderTestHarn
   @Override
   protected Properties getMetaProps() {
     Properties metaProps =  super.getMetaProps();
-    metaProps.setProperty(HoodieTableConfig.RECORD_MERGE_MODE.key(), RecordMergeMode.OVERWRITE_WITH_LATEST.name());
-    metaProps.setProperty(HoodieTableConfig.RECORD_MERGE_STRATEGY_ID.key(), HoodieRecordMerger.OVERWRITE_MERGE_STRATEGY_UUID);
+    metaProps.setProperty(HoodieTableConfig.RECORD_MERGE_MODE.key(), RecordMergeMode.COMMIT_TIME_ORDERING.name());
+    metaProps.setProperty(HoodieTableConfig.RECORD_MERGE_STRATEGY_ID.key(), HoodieRecordMerger.COMMIT_TIME_BASED_MERGE_STRATEGY_UUID);
     metaProps.setProperty(HoodieTableConfig.PAYLOAD_CLASS_NAME.key(), OverwriteWithLatestAvroPayload.class.getName());
     return metaProps;
   }
@@ -71,7 +71,7 @@ public static void setUp() throws IOException {
     readerContext = new HoodieTestReaderContext(
         Option.of(merger),
         Option.of(OverwriteWithLatestAvroPayload.class.getName()));
-    properties.setProperty("hoodie.write.record.merge.mode", RecordMergeMode.OVERWRITE_WITH_LATEST.name());
+    properties.setProperty("hoodie.write.record.merge.mode", RecordMergeMode.COMMIT_TIME_ORDERING.name());
 
     // -------------------------------------------------------------
     // The test logic is as follows: