[KYUUBI #6884][FOLLOWUP] Fix internal kyuubi instance ping failure

### Why are the changes needed?

To fix failure:
```
:2025-12-07 13:40:40.747 ERROR [KyuubiRestFrontendService-134] org.apache.kyuubi.server.api.v1.InternalRestClient: (Ping to internal Kyuubi instance:{}  failed,kyuubistaging-1.kyuubistaginghl.kyuubi-prod.svc.166.tess.io:10011,java.lang.IllegalArgumentException: requirement failed: The auth user shall be not null)
```

In this PR, it will use the authUser and client ipAddress for internal ping.

### How was this patch tested?

Integration testing.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #7264 from turboFei/ping_able.

Closes #6884

198dc61 [Wang, Fei] debug

Authored-by: Wang, Fei <fwang12@ebay.com>
Signed-off-by: Wang, Fei <fwang12@ebay.com>

Author: turboFei

kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/BaseRestApi.java

@@ -17,6 +17,8 @@
 
 package org.apache.kyuubi.client;
 
+import java.util.Collections;
+import java.util.Map;
 import org.apache.kyuubi.client.api.v1.dto.VersionInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -32,7 +34,11 @@ public BaseRestApi(KyuubiRestClient client) {
   }
 
   public String ping() {
-    return this.getClient().get("ping", null, client.getAuthHeader());
+    return ping(Collections.emptyMap());
+  }
+
+  public String ping(Map<String, String> headers) {
+    return this.getClient().get("ping", null, client.getAuthHeader(), headers);
   }
 
   public VersionInfo getVersionInfo() {

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala

@@ -598,7 +598,7 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
       throw new IllegalStateException(s"KyuubiInstance is alive: $kyuubiInstance")
     }
     val internalRestClient = getInternalRestClient(kyuubiInstance)
-    if (!Utils.isTesting && internalRestClient.pingAble()) {
+    if (!Utils.isTesting && internalRestClient.pingAble(userName, ipAddress)) {
       throw new IllegalStateException(s"KyuubiInstance is alive: $kyuubiInstance")
     }
     try {

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/InternalRestClient.scala

@@ -20,8 +20,9 @@ package org.apache.kyuubi.server.api.v1
 import java.util.Base64
 
 import scala.collection.JavaConverters._
-import scala.util.Try
+import scala.util.{Failure, Success, Try}
 
+import org.apache.kyuubi.Logging
 import org.apache.kyuubi.client.{BaseRestApi, BatchRestApi, KyuubiRestClient}
 import org.apache.kyuubi.client.api.v1.dto.{Batch, CloseBatchResponse, OperationLog}
 import org.apache.kyuubi.client.auth.AuthHeaderGenerator
@@ -46,7 +47,7 @@ class InternalRestClient(
     connectTimeout: Int,
     securityEnabled: Boolean,
     requestMaxAttempts: Int,
-    requestAttemptWait: Int) {
+    requestAttemptWait: Int) extends Logging {
   if (securityEnabled) {
     require(
       InternalSecurityAccessor.get() != null,
@@ -56,7 +57,16 @@ class InternalRestClient(
   private val internalBatchRestApi = new BatchRestApi(initKyuubiRestClient())
   private val internalBaseRestApi = new BaseRestApi(initKyuubiRestClient())
 
-  def pingAble(): Boolean = Try(internalBaseRestApi.ping()).isSuccess
+  def pingAble(user: String, clientIp: String): Boolean = withAuthUser(user) {
+    Try {
+      internalBaseRestApi.ping(Map(proxyClientIpHeader -> clientIp).asJava)
+    } match {
+      case Success(_) => true
+      case Failure(e) =>
+        error(s"Ping to Kyuubi instance $kyuubiInstance failed", e)
+        false
+    }
+  }
 
   def getBatch(user: String, clientIp: String, batchId: String): Batch = {
     withAuthUser(user) {