[Kernel] Remove Coordinated Commits from public API (delta-io#3938)

- [ ] Spark
- [ ] Standalone
- [ ] Flink
- [X] Kernel
- [ ] Other (fill in here)

We are re-thinking the design of the Coordinated Commits table feature
(currently still in RFC). Thus, we should remove it from the public
Kernel API for Delta 3.3 release.

To summarize the changes of this PR

- I remove `getCommitCoordinatorClientHandler` from the `Engine`
interface
- I move various previously `public` CC interfaces and classes to be
`internal` now
- `SnapshotImpl::getTableCommitCoordinatorClientHandlerOpt` is hardcoded
to return an empty optional
- Delete failing test suites and unapplicable utils

Existing CI tests.

We remove coordinated commits from the public kernel API.

kernel/kernel-api/src/main/java/io/delta/kernel/engine/Engine.java

@@ -17,7 +17,6 @@
 package io.delta.kernel.engine;
 
 import io.delta.kernel.annotation.Evolving;
-import java.util.Map;
 
 /**
  * Interface encapsulating all clients needed by the Delta Kernel in order to read the Delta table.
@@ -55,25 +54,4 @@ public interface Engine {
    * @return An implementation of {@link ParquetHandler}.
    */
   ParquetHandler getParquetHandler();
-
-  /**
-   * Retrieves a {@link CommitCoordinatorClientHandler} for the specified commit coordinator client.
-   *
-   * <p>{@link CommitCoordinatorClientHandler} helps Kernel perform commits to a table which is
-   * owned by a commit coordinator.
-   *
-   * @see <a
-   *     href="https://github.com/delta-io/delta/blob/master/protocol_rfcs/managed-commits.md#sample-commit-owner-api">Coordinated
-   *     commit protocol table feature</a>.
-   *     <p>This method creates and returns an implementation of {@link
-   *     CommitCoordinatorClientHandler} based on the provided name and configuration of the
-   *     underlying commit coordinator client.
-   * @param name The identifier or name of the underlying commit coordinator client
-   * @param conf The configuration settings for the underlying commit coordinator client
-   * @return An implementation of {@link CommitCoordinatorClientHandler} configured for the
-   *     specified client
-   * @since 3.3.0
-   */
-  CommitCoordinatorClientHandler getCommitCoordinatorClientHandler(
-      String name, Map<String, String> conf);
 }

kernel/kernel-api/src/main/java/io/delta/kernel/internal/SnapshotImpl.java

@@ -20,7 +20,6 @@
 
 import io.delta.kernel.ScanBuilder;
 import io.delta.kernel.Snapshot;
-import io.delta.kernel.engine.CommitCoordinatorClientHandler;
 import io.delta.kernel.engine.Engine;
 import io.delta.kernel.internal.actions.CommitInfo;
 import io.delta.kernel.internal.actions.DomainMetadata;
@@ -180,18 +179,6 @@ public Optional<Long> getLatestTransactionVersion(Engine engine, String applicat
    */
   public Optional<TableCommitCoordinatorClientHandler> getTableCommitCoordinatorClientHandlerOpt(
       Engine engine) {
-    return COORDINATED_COMMITS_COORDINATOR_NAME
-        .fromMetadata(metadata)
-        .map(
-            commitCoordinatorStr -> {
-              CommitCoordinatorClientHandler handler =
-                  engine.getCommitCoordinatorClientHandler(
-                      commitCoordinatorStr,
-                      COORDINATED_COMMITS_COORDINATOR_CONF.fromMetadata(metadata));
-              return new TableCommitCoordinatorClientHandler(
-                  handler,
-                  logPath.toString(),
-                  COORDINATED_COMMITS_TABLE_CONF.fromMetadata(metadata));
-            });
+    return Optional.empty();
   }
 }

kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/Commit.java → kernel/kernel-api/src/main/java/io/delta/kernel/internal/coordinatedcommits/Commit.java

@@ -14,16 +14,16 @@
  * limitations under the License.
  */
 
-package io.delta.kernel.engine.coordinatedcommits;
+package io.delta.kernel.internal.coordinatedcommits;
 
 import io.delta.kernel.annotation.Evolving;
 import io.delta.kernel.utils.FileStatus;
 
 /**
  * Representation of a commit file. It contains the version of the commit, the file status of the
  * commit, and the timestamp of the commit. This is used when we want to get the commit information
- * from the {@link io.delta.kernel.engine.CommitCoordinatorClientHandler#commit} and {@link
- * io.delta.kernel.engine.CommitCoordinatorClientHandler#getCommits} APIs.
+ * from the {@link CommitCoordinatorClientHandler#commit} and {@link
+ * CommitCoordinatorClientHandler#getCommits} APIs.
  *
  * @since 3.3.0
  */

kernel/kernel-api/src/main/java/io/delta/kernel/engine/CommitCoordinatorClientHandler.java → kernel/kernel-api/src/main/java/io/delta/kernel/internal/coordinatedcommits/CommitCoordinatorClientHandler.java

@@ -14,13 +14,12 @@
  * limitations under the License.
  */
 
-package io.delta.kernel.engine;
+package io.delta.kernel.internal.coordinatedcommits;
 
 import io.delta.kernel.annotation.Evolving;
 import io.delta.kernel.data.Row;
-import io.delta.kernel.engine.coordinatedcommits.*;
-import io.delta.kernel.engine.coordinatedcommits.actions.AbstractMetadata;
-import io.delta.kernel.engine.coordinatedcommits.actions.AbstractProtocol;
+import io.delta.kernel.internal.coordinatedcommits.actions.AbstractMetadata;
+import io.delta.kernel.internal.coordinatedcommits.actions.AbstractProtocol;
 import io.delta.kernel.utils.CloseableIterator;
 import java.io.IOException;
 import java.util.Map;

kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/CommitFailedException.java → kernel/kernel-api/src/main/java/io/delta/kernel/internal/coordinatedcommits/CommitFailedException.java

@@ -14,12 +14,12 @@
  * limitations under the License.
  */
 
-package io.delta.kernel.engine.coordinatedcommits;
+package io.delta.kernel.internal.coordinatedcommits;
 
 import io.delta.kernel.annotation.Evolving;
 
 /**
- * Exception raised by {@link io.delta.kernel.engine.CommitCoordinatorClientHandler#commit}
+ * Exception raised by {@link CommitCoordinatorClientHandler#commit}
  *
  * <pre>
  *  | retryable | conflict  | meaning                                                         |

kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/CommitResponse.java → kernel/kernel-api/src/main/java/io/delta/kernel/internal/coordinatedcommits/CommitResponse.java

@@ -14,12 +14,12 @@
  * limitations under the License.
  */
 
-package io.delta.kernel.engine.coordinatedcommits;
+package io.delta.kernel.internal.coordinatedcommits;
 
 import io.delta.kernel.annotation.Evolving;
 
 /**
- * Response container for {@link io.delta.kernel.engine.CommitCoordinatorClientHandler#commit}.
+ * Response container for {@link CommitCoordinatorClientHandler#commit}.
  *
  * @since 3.3.0
  */

kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/GetCommitsResponse.java → kernel/kernel-api/src/main/java/io/delta/kernel/internal/coordinatedcommits/GetCommitsResponse.java

@@ -14,16 +14,15 @@
  * limitations under the License.
  */
 
-package io.delta.kernel.engine.coordinatedcommits;
+package io.delta.kernel.internal.coordinatedcommits;
 
 import io.delta.kernel.annotation.Evolving;
 import java.util.List;
 import java.util.Map;
 
 /**
- * Response container for {@link io.delta.kernel.engine.CommitCoordinatorClientHandler#getCommits(
- * String, Map, Long, Long)}. Holds all the commits that have not been backfilled as per the commit
- * coordinator.
+ * Response container for {@link CommitCoordinatorClientHandler#getCommits( String, Map, Long,
+ * Long)}. Holds all the commits that have not been backfilled as per the commit coordinator.
  *
  * @since 3.3.0
  */

kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/UpdatedActions.java → kernel/kernel-api/src/main/java/io/delta/kernel/internal/coordinatedcommits/UpdatedActions.java

@@ -14,16 +14,16 @@
  * limitations under the License.
  */
 
-package io.delta.kernel.engine.coordinatedcommits;
+package io.delta.kernel.internal.coordinatedcommits;
 
 import io.delta.kernel.annotation.Evolving;
-import io.delta.kernel.engine.coordinatedcommits.actions.AbstractCommitInfo;
-import io.delta.kernel.engine.coordinatedcommits.actions.AbstractMetadata;
-import io.delta.kernel.engine.coordinatedcommits.actions.AbstractProtocol;
+import io.delta.kernel.internal.coordinatedcommits.actions.AbstractCommitInfo;
+import io.delta.kernel.internal.coordinatedcommits.actions.AbstractMetadata;
+import io.delta.kernel.internal.coordinatedcommits.actions.AbstractProtocol;
 
 /**
- * A container class to inform the {@link io.delta.kernel.engine.CommitCoordinatorClientHandler}
- * about any changes in Protocol/Metadata
+ * A container class to inform the {@link CommitCoordinatorClientHandler} about any changes in
+ * Protocol/Metadata
  *
  * @since 3.3.0
  */

kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/actions/AbstractCommitInfo.java → kernel/kernel-api/src/main/java/io/delta/kernel/internal/coordinatedcommits/actions/AbstractCommitInfo.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.delta.kernel.engine.coordinatedcommits.actions;
+package io.delta.kernel.internal.coordinatedcommits.actions;
 
 import io.delta.kernel.annotation.Evolving;
 

kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/actions/AbstractMetadata.java → kernel/kernel-api/src/main/java/io/delta/kernel/internal/coordinatedcommits/actions/AbstractMetadata.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.delta.kernel.engine.coordinatedcommits.actions;
+package io.delta.kernel.internal.coordinatedcommits.actions;
 
 import io.delta.kernel.annotation.Evolving;
 import java.util.List;

kernel/kernel-api/src/main/java/io/delta/kernel/engine/coordinatedcommits/actions/AbstractProtocol.java → kernel/kernel-api/src/main/java/io/delta/kernel/internal/coordinatedcommits/actions/AbstractProtocol.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package io.delta.kernel.engine.coordinatedcommits.actions;
+package io.delta.kernel.internal.coordinatedcommits.actions;
 
 import io.delta.kernel.annotation.Evolving;
 import java.util.Set;

kernel/kernel-api/src/main/java/io/delta/kernel/internal/snapshot/SnapshotManager.java

@@ -29,13 +29,13 @@
 
 import io.delta.kernel.*;
 import io.delta.kernel.engine.Engine;
-import io.delta.kernel.engine.coordinatedcommits.Commit;
 import io.delta.kernel.exceptions.CheckpointAlreadyExistsException;
 import io.delta.kernel.exceptions.InvalidTableException;
 import io.delta.kernel.exceptions.TableNotFoundException;
 import io.delta.kernel.internal.*;
 import io.delta.kernel.internal.actions.Metadata;
 import io.delta.kernel.internal.checkpoints.*;
+import io.delta.kernel.internal.coordinatedcommits.Commit;
 import io.delta.kernel.internal.fs.Path;
 import io.delta.kernel.internal.lang.ListUtils;
 import io.delta.kernel.internal.replay.CreateCheckpointIterator;

kernel/kernel-api/src/main/java/io/delta/kernel/internal/snapshot/TableCommitCoordinatorClientHandler.java

@@ -17,11 +17,11 @@
 package io.delta.kernel.internal.snapshot;
 
 import io.delta.kernel.data.Row;
-import io.delta.kernel.engine.CommitCoordinatorClientHandler;
-import io.delta.kernel.engine.coordinatedcommits.CommitFailedException;
-import io.delta.kernel.engine.coordinatedcommits.CommitResponse;
-import io.delta.kernel.engine.coordinatedcommits.GetCommitsResponse;
-import io.delta.kernel.engine.coordinatedcommits.UpdatedActions;
+import io.delta.kernel.internal.coordinatedcommits.CommitCoordinatorClientHandler;
+import io.delta.kernel.internal.coordinatedcommits.CommitFailedException;
+import io.delta.kernel.internal.coordinatedcommits.CommitResponse;
+import io.delta.kernel.internal.coordinatedcommits.GetCommitsResponse;
+import io.delta.kernel.internal.coordinatedcommits.UpdatedActions;
 import io.delta.kernel.utils.CloseableIterator;
 import java.io.IOException;
 import java.util.Map;

kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/CoordinatedCommitsUtils.java

@@ -15,12 +15,12 @@
  */
 package io.delta.kernel.internal.util;
 
-import io.delta.kernel.engine.coordinatedcommits.actions.AbstractCommitInfo;
-import io.delta.kernel.engine.coordinatedcommits.actions.AbstractMetadata;
-import io.delta.kernel.engine.coordinatedcommits.actions.AbstractProtocol;
 import io.delta.kernel.internal.actions.CommitInfo;
 import io.delta.kernel.internal.actions.Metadata;
 import io.delta.kernel.internal.actions.Protocol;
+import io.delta.kernel.internal.coordinatedcommits.actions.AbstractCommitInfo;
+import io.delta.kernel.internal.coordinatedcommits.actions.AbstractMetadata;
+import io.delta.kernel.internal.coordinatedcommits.actions.AbstractProtocol;
 import java.util.*;
 
 public class CoordinatedCommitsUtils {

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/SnapshotManagerSuite.scala

@@ -20,12 +20,11 @@ import java.{lang => javaLang}
 import scala.collection.JavaConverters._
 import scala.reflect.ClassTag
 import io.delta.kernel.data.{ColumnarBatch, ColumnVector}
-import io.delta.kernel.engine.CommitCoordinatorClientHandler
-import io.delta.kernel.engine.coordinatedcommits.{Commit, CommitResponse, GetCommitsResponse}
 import io.delta.kernel.exceptions.InvalidTableException
 import io.delta.kernel.expressions.Predicate
 import io.delta.kernel.internal.actions.CommitInfo
 import io.delta.kernel.internal.checkpoints.{CheckpointInstance, SidecarFile}
+import io.delta.kernel.internal.coordinatedcommits.{Commit, CommitCoordinatorClientHandler, CommitResponse, GetCommitsResponse}
 import io.delta.kernel.internal.fs.Path
 import io.delta.kernel.internal.snapshot.{LogSegment, SnapshotManager, TableCommitCoordinatorClientHandler}
 import io.delta.kernel.internal.util.{FileNames, Utils}

kernel/kernel-api/src/test/scala/io/delta/kernel/test/MockEngineUtils.scala

@@ -18,6 +18,7 @@ package io.delta.kernel.test
 import io.delta.kernel.engine._
 import io.delta.kernel.data.{ColumnVector, ColumnarBatch, FilteredColumnarBatch, Row}
 import io.delta.kernel.expressions.{Column, Expression, ExpressionEvaluator, Predicate, PredicateEvaluator}
+import io.delta.kernel.internal.coordinatedcommits.CommitCoordinatorClientHandler
 import io.delta.kernel.types.{DataType, StructType}
 import io.delta.kernel.utils.{CloseableIterator, DataFileStatus, FileStatus}
 
@@ -70,11 +71,6 @@ trait MockEngineUtils {
       override def getParquetHandler: ParquetHandler =
         Option(parquetHandler).getOrElse(
           throw new UnsupportedOperationException("not supported in this test suite"))
-
-      override def getCommitCoordinatorClientHandler(name: String, conf: util.Map[String, String]):
-      CommitCoordinatorClientHandler =
-        Option(commitCoordinatorClientHandler).getOrElse(
-          throw new UnsupportedOperationException("not supported in this test suite"))
     }
   }
 }