[#4973] improvement(client-java/server): Clear CallerContext in Jav…

…a Client and Server when call `getFileLocation` finally (#4974)

### What changes were proposed in this pull request?

Currently, after the client completes the `getFileLocation` request and
the server completes the `getFileLocation` response, the Thread Local
`CallerContext` is not cleaned up. This may cause the next request to
incorrectly report the same information repeatedly when the thread is
reused. We should eventually clean up the `CallerContext`.

### Why are the changes needed?

Fix: #4973 

### How was this patch tested?

Add UTs.

clients/client-java/src/main/java/org/apache/gravitino/client/FilesetCatalog.java

@@ -244,20 +244,25 @@ public String getFileLocation(NameIdentifier ident, String subPath)
     checkFilesetNameIdentifier(ident);
     Namespace fullNamespace = getFilesetFullNamespace(ident.namespace());
 
-    CallerContext callerContext = CallerContext.CallerContextHolder.get();
-
-    Map<String, String> params = new HashMap<>();
-    params.put("sub_path", RESTUtils.encodeString(subPath));
-    FileLocationResponse resp =
-        restClient.get(
-            formatFileLocationRequestPath(fullNamespace, ident.name()),
-            params,
-            FileLocationResponse.class,
-            callerContext != null ? callerContext.context() : Collections.emptyMap(),
-            ErrorHandlers.filesetErrorHandler());
-    resp.validate();
-
-    return resp.getFileLocation();
+    try {
+      CallerContext callerContext = CallerContext.CallerContextHolder.get();
+
+      Map<String, String> params = new HashMap<>();
+      params.put("sub_path", RESTUtils.encodeString(subPath));
+      FileLocationResponse resp =
+          restClient.get(
+              formatFileLocationRequestPath(fullNamespace, ident.name()),
+              params,
+              FileLocationResponse.class,
+              callerContext != null ? callerContext.context() : Collections.emptyMap(),
+              ErrorHandlers.filesetErrorHandler());
+      resp.validate();
+
+      return resp.getFileLocation();
+    } finally {
+      // Clear the caller context
+      CallerContext.CallerContextHolder.remove();
+    }
   }
 
   @VisibleForTesting

clients/client-java/src/test/java/org/apache/gravitino/client/TestFilesetCatalog.java

@@ -534,6 +534,8 @@ public void testCallerContextToHeader() throws JsonProcessingException {
             NameIdentifier.of(fileset.namespace().level(2), fileset.name()), mockSubPath);
     Assertions.assertEquals(FilesetDataOperation.GET_FILE_STATUS.name(), dataOperation.get());
     Assertions.assertEquals(InternalClientType.HADOOP_GVFS.name(), internalClientType.get());
+    // the caller context should be cleared after `getFileLocation`
+    Assertions.assertNull(CallerContext.CallerContextHolder.get());
   }
 
   private FilesetDTO mockFilesetDTO(

server/src/main/java/org/apache/gravitino/server/web/rest/FilesetOperations.java

@@ -291,6 +291,9 @@ public Response getFileLocation(
           });
     } catch (Exception e) {
       return ExceptionHandlers.handleFilesetException(OperationType.GET, fileset, schema, e);
+    } finally {
+      // Clear the caller context
+      CallerContext.CallerContextHolder.remove();
     }
   }
 }

server/src/test/java/org/apache/gravitino/server/web/rest/TestFilesetOperations.java

@@ -554,6 +554,18 @@ public void testGetFileLocation() {
     } finally {
       CallerContext.CallerContextHolder.remove();
     }
+
+    // test `getFileLocation` clear caller context finally
+    Map<String, String> testContextMap = Maps.newHashMap();
+    testContextMap.put("k", "v");
+    CallerContext context = CallerContext.builder().withContext(testContextMap).build();
+    CallerContext.CallerContextHolder.set(context);
+    FilesetOperations mockOperations = Mockito.mock(FilesetOperations.class);
+    Mockito.when(mockOperations.getFileLocation(any(), any(), any(), any(), any()))
+        .thenCallRealMethod();
+    mockOperations.getFileLocation(
+        "test_metalake", "test_catalog", "test_schema", "fileset4", "/test");
+    Assertions.assertNull(CallerContext.CallerContextHolder.get());
   }
 
   private void assertUpdateFileset(FilesetUpdatesRequest req, Fileset updatedFileset) {