[HUDI-5082] Improve the cdc log file name format (apache#7042)

Author: YannByron

hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java

@@ -80,7 +80,7 @@ public class FSUtils {
   // Log files are of this pattern - .b5068208-e1a4-11e6-bf01-fe55135034f3_20170101134598.log.1_1-0-1
   // Archive log files are of this pattern - .commits_.archive.1_1-0-1
   private static final Pattern LOG_FILE_PATTERN =
-      Pattern.compile("\\.(.+)_(.*)\\.(.+)\\.(\\d+)(_((\\d+)-(\\d+)-(\\d+))(-cdc)?)?");
+      Pattern.compile("\\.(.+)_(.*)\\.(.+)\\.(\\d+)(_((\\d+)-(\\d+)-(\\d+))(.cdc)?)?");
   private static final String LOG_FILE_PREFIX = ".";
   private static final int MAX_ATTEMPTS_RECOVER_LEASE = 10;
   private static final long MIN_CLEAN_TO_KEEP = 10;

hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCUtils.java

@@ -34,7 +34,7 @@
  */
 public class HoodieCDCUtils {
 
-  public static final String CDC_LOGFILE_SUFFIX = "-cdc";
+  public static final String CDC_LOGFILE_SUFFIX = ".cdc";
 
   /* the `op` column represents how a record is changed. */
   public static final String CDC_OPERATION_TYPE = "op";

hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/cdc/HoodieCDCTestBase.scala

@@ -21,7 +21,7 @@ import org.apache.hudi.DataSourceReadOptions._
 import org.apache.hudi.DataSourceWriteOptions._
 import org.apache.hudi.common.config.HoodieMetadataConfig
 import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieKey, HoodieLogFile, HoodieRecord}
-import org.apache.hudi.common.table.cdc.HoodieCDCOperation
+import org.apache.hudi.common.table.cdc.{HoodieCDCOperation, HoodieCDCUtils}
 import org.apache.hudi.common.table.HoodieTableConfig
 import org.apache.hudi.common.table.log.HoodieLogFormat
 import org.apache.hudi.common.table.log.block.HoodieDataBlock
@@ -101,7 +101,8 @@ abstract class HoodieCDCTestBase extends HoodieClientTestBase {
     val hoodieWriteStats = commitMetadata.getWriteStats.asScala
     hoodieWriteStats.exists { hoodieWriteStat =>
       val cdcPaths = hoodieWriteStat.getCdcStats
-      cdcPaths != null && cdcPaths.nonEmpty
+      cdcPaths != null && cdcPaths.nonEmpty &&
+        cdcPaths.keySet().asScala.forall(_.endsWith(HoodieCDCUtils.CDC_LOGFILE_SUFFIX))
     }
   }