refine call context and ut after merge

Author: danielhumanmod

quarkus/service/src/test/java/org/apache/polaris/service/quarkus/task/BatchFileCleanupTaskHandlerTest.java

@@ -69,7 +69,7 @@ private TaskFileIOSupplier buildTaskFileIOSupplier(FileIO fileIO) {
         new FileIOFactory() {
           @Override
           public FileIO loadFileIO(
-              @NotNull RealmContext realmContext,
+              @NotNull CallContext callContext,
               @NotNull String ioImplClassName,
               @NotNull Map<String, String> properties,
               @NotNull TableIdentifier identifier,

quarkus/service/src/test/java/org/apache/polaris/service/quarkus/task/TableCleanupTaskHandlerTest.java

@@ -149,9 +149,7 @@ public void testTableCleanup() throws IOException {
                 assertThat(taskEntity)
                     .returns(PolarisEntityType.TASK.getCode(), PolarisBaseEntity::getTypeCode)
                     .extracting(TaskEntity::of)
-                    .returns(
-                        AsyncTaskType.MANIFEST_FILE_CLEANUP,
-                        taskEntity1 -> taskEntity1.getTaskType())
+                    .returns(AsyncTaskType.MANIFEST_FILE_CLEANUP, TaskEntity::getTaskType)
                     .returns(
                         new ManifestFileCleanupTaskHandler.ManifestCleanupTask(
                             tableIdentifier,
@@ -167,7 +165,8 @@ public void testTableCleanup() throws IOException {
                         AsyncTaskType.BATCH_FILE_CLEANUP, taskEntity2 -> taskEntity2.getTaskType())
                     .returns(
                         new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
-                            tableIdentifier, List.of(statisticsFile.path())),
+                            tableIdentifier,
+                            List.of(snapshot.manifestListLocation(), statisticsFile.path())),
                         entity ->
                             entity.readData(
                                 BatchFileCleanupTaskHandler.BatchFileCleanupTask.class)));
@@ -291,35 +290,29 @@ public void close() {
                 assertThat(taskEntity)
                     .returns(PolarisEntityType.TASK.getCode(), PolarisBaseEntity::getTypeCode)
                     .extracting(TaskEntity::of)
+                    .returns(AsyncTaskType.BATCH_FILE_CLEANUP, TaskEntity::getTaskType)
                     .returns(
-                        AsyncTaskType.METADATA_FILE_BATCH_CLEANUP,
-                        taskEntity1 -> taskEntity1.getTaskType())
-                    .returns(
-                        new ManifestFileCleanupTaskHandler.ManifestCleanupTask(
+                        new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
                             tableIdentifier, List.of(snapshot.manifestListLocation())),
                         entity ->
                             entity.readData(
-                                ManifestFileCleanupTaskHandler.ManifestCleanupTask.class)),
+                                BatchFileCleanupTaskHandler.BatchFileCleanupTask.class)),
             taskEntity ->
                 assertThat(taskEntity)
                     .returns(PolarisEntityType.TASK.getCode(), PolarisBaseEntity::getTypeCode)
                     .extracting(TaskEntity::of)
+                    .returns(AsyncTaskType.BATCH_FILE_CLEANUP, TaskEntity::getTaskType)
                     .returns(
-                        AsyncTaskType.METADATA_FILE_BATCH_CLEANUP,
-                        taskEntity2 -> taskEntity2.getTaskType())
-                    .returns(
-                        new ManifestFileCleanupTaskHandler.ManifestCleanupTask(
+                        new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
                             tableIdentifier, List.of(snapshot.manifestListLocation())),
                         entity ->
                             entity.readData(
-                                ManifestFileCleanupTaskHandler.ManifestCleanupTask.class)),
+                                BatchFileCleanupTaskHandler.BatchFileCleanupTask.class)),
             taskEntity ->
                 assertThat(taskEntity)
                     .returns(PolarisEntityType.TASK.getCode(), PolarisBaseEntity::getTypeCode)
                     .extracting(TaskEntity::of)
-                    .returns(
-                        AsyncTaskType.MANIFEST_FILE_CLEANUP,
-                        taskEntity3 -> taskEntity3.getTaskType())
+                    .returns(AsyncTaskType.MANIFEST_FILE_CLEANUP, TaskEntity::getTaskType)
                     .returns(
                         new ManifestFileCleanupTaskHandler.ManifestCleanupTask(
                             tableIdentifier,

service/common/src/main/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandler.java

@@ -53,7 +53,7 @@ public boolean handleTask(TaskEntity task, CallContext callContext) {
     BatchFileCleanupTask cleanupTask = task.readData(BatchFileCleanupTask.class);
     TableIdentifier tableId = cleanupTask.tableId();
     List<String> batchFiles = cleanupTask.batchFiles();
-    try (FileIO authorizedFileIO = fileIOSupplier.apply(task, callContext.getRealmContext())) {
+    try (FileIO authorizedFileIO = fileIOSupplier.apply(task, callContext)) {
       List<String> validFiles =
           batchFiles.stream().filter(file -> TaskUtils.exists(file, authorizedFileIO)).toList();
       if (validFiles.isEmpty()) {

service/common/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java

@@ -39,8 +39,7 @@ public abstract class FileCleanupTaskHandler implements TaskHandler {
   public static final int FILE_DELETION_RETRY_MILLIS = 100;
   public final TaskFileIOSupplier fileIOSupplier;
   public final ExecutorService executorService;
-  private static final Logger LOGGER =
-      LoggerFactory.getLogger(ManifestFileCleanupTaskHandler.class);
+  private static final Logger LOGGER = LoggerFactory.getLogger(FileCleanupTaskHandler.class);
 
   public FileCleanupTaskHandler(
       TaskFileIOSupplier fileIOSupplier, ExecutorService executorService) {