[Spark][Kernel] Remove duplicate CoordinatedCommitsUtils.java from Dy…

…namoDBCommitCoordinator (#3653)

#### Which Delta project/connector is this regarding?

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [x] Kernel
- [ ] Other

## Description

- Remove duplicate CoordinatedCommitsUtils.java from
DynamoDBCommitCoordinator package.
- Move the common utils to non-test package.

## How was this patch tested?

Compile

## Does this PR introduce _any_ user-facing changes?

No

spark/src/main/java/io/delta/dynamodbcommitcoordinator/DynamoDBCommitCoordinatorClient.java

@@ -416,7 +416,7 @@ private GetCommitsResultInternal getCommitsImpl(
                 Long.parseLong(item.get(DynamoDBTableEntryConstants.TABLE_LATEST_VERSION).getN());
         AttributeValue allStoredCommits = item.get(DynamoDBTableEntryConstants.COMMITS);
         ArrayList<Commit> commits = new ArrayList<>();
-        Path unbackfilledCommitsPath = new Path(logPath, CoordinatedCommitsUtils.COMMIT_SUBDIR);
+        Path unbackfilledCommitsPath = CoordinatedCommitsUtils.commitDirPath(logPath);
         for(AttributeValue attr: allStoredCommits.getL()) {
             java.util.Map<String, AttributeValue> commitMap = attr.getM();
             long commitVersion =

storage/src/test/scala/io/delta/storage/commit/CoordinatedCommitsUtils.java → storage/src/main/java/io/delta/storage/commit/CoordinatedCommitsUtils.java

@@ -33,12 +33,24 @@ public class CoordinatedCommitsUtils {
     private CoordinatedCommitsUtils() {}
 
     /** The subdirectory in which to store the unbackfilled commit files. */
-    final static String COMMIT_SUBDIR = "_commits";
+    private static final String COMMIT_SUBDIR = "_commits";
 
-    /** The configuration key for the coordinated commits owner. */
-    private static final String COORDINATED_COMMITS_COORDINATOR_CONF_KEY =
+    /** The configuration key for the coordinated commits owner name. */
+    private static final String COORDINATED_COMMITS_COORDINATOR_NAME_KEY =
             "delta.coordinatedCommits.commitCoordinator-preview";
 
+    /**
+     * Creates a new unbackfilled delta file path for the given commit version.
+     * The path is of the form `tablePath/_delta_log/_commits/00000000000000000001.uuid.json`.
+     */
+    public static Path generateUnbackfilledDeltaFilePath(
+            Path logPath,
+            long version) {
+        String uuid = UUID.randomUUID().toString();
+        Path basePath = new Path(logPath, COMMIT_SUBDIR);
+        return new Path(basePath, String.format("%020d.%s.json", version, uuid));
+    }
+
     /**
      * Returns the path to the backfilled delta file for the given commit version.
      * The path is of the form `tablePath/_delta_log/00000000000000000001.json`.
@@ -108,10 +120,10 @@ public static Path commitDirPath(Path logPath) {
         return new Path(logPath, COMMIT_SUBDIR);
     }
 
-    private static String getCoordinator(AbstractMetadata metadata) {
+    public static String getCoordinator(AbstractMetadata metadata) {
         String coordinator = metadata
                 .getConfiguration()
-                .get(COORDINATED_COMMITS_COORDINATOR_CONF_KEY);
+                .get(COORDINATED_COMMITS_COORDINATOR_NAME_KEY);
         return coordinator != null ? coordinator : "";
     }
 }

storage/src/test/scala/io/delta/storage/commit/InMemoryCommitCoordinator.scala

@@ -22,6 +22,7 @@ import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.locks.ReentrantReadWriteLock
 
 import io.delta.storage.LogStore
+import io.delta.storage.commit.CoordinatedCommitsUtils
 import io.delta.storage.commit.actions.AbstractMetadata
 import io.delta.storage.commit.actions.AbstractProtocol
 import org.apache.hadoop.conf.Configuration