Add (and process at the top) SDK specific exceptions to distinguish errors

src/main/java/com/rawlabs/das/sdk/DASSdkPermissionDeniedException.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2025 RAW Labs S.A.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.txt.
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0, included in the file
+ * licenses/APL.txt.
+ */
+
+package com.rawlabs.das.sdk;
+
+/**
+ * DASSdkPermissionDeniedException is thrown by DAS SDK methods when a permission denied error is to
+ * be reported to a user (e.g. missing required permissions, etc.).
+ */
+public class DASSdkPermissionDeniedException extends RuntimeException {
+  public DASSdkPermissionDeniedException(String message) {
+    super(message);
+  }
+
+  public DASSdkPermissionDeniedException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

src/main/java/com/rawlabs/das/sdk/DASSdkUnauthenticatedException.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2025 RAW Labs S.A.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.txt.
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0, included in the file
+ * licenses/APL.txt.
+ */
+
+package com.rawlabs.das.sdk;
+
+/**
+ * DASSdkUnauthenticatedException is thrown by DAS SDK methods when an unauthenticated error is to
+ * be reported to a user (e.g. missing authentication token, invalid credentials, etc.).
+ */
+public class DASSdkUnauthenticatedException extends RuntimeException {
+  public DASSdkUnauthenticatedException(String message) {
+    super(message);
+  }
+
+  public DASSdkUnauthenticatedException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

src/main/scala/com/rawlabs/das/server/grpc/RegistrationServiceGrpcImpl.scala

@@ -14,11 +14,18 @@ package com.rawlabs.das.server.grpc
 
 import scala.jdk.CollectionConverters._
 
+import com.rawlabs.das.sdk.{
+  DASSdkInvalidArgumentException,
+  DASSdkPermissionDeniedException,
+  DASSdkUnauthenticatedException,
+  DASSdkUnsupportedException
+}
 import com.rawlabs.das.server.manager.DASSdkManager
 import com.rawlabs.protocol.das.v1.common.DASId
 import com.rawlabs.protocol.das.v1.services._
 import com.typesafe.scalalogging.StrictLogging
 
+import io.grpc.Status
 import io.grpc.stub.StreamObserver
 
 /**
@@ -38,13 +45,32 @@ class RegistrationServiceGrpcImpl(dasSdkManager: DASSdkManager)
    */
   override def register(request: RegisterRequest, responseObserver: StreamObserver[RegisterResponse]): Unit = {
     logger.debug(s"Registering DAS with type: ${request.getDefinition.getType}")
-    val dasId = dasSdkManager.registerDAS(
-      request.getDefinition.getType,
-      request.getDefinition.getOptionsMap.asScala.toMap,
-      maybeDasId = if (request.hasId) Some(request.getId) else None)
-    responseObserver.onNext(dasId)
-    responseObserver.onCompleted()
-    logger.debug(s"DAS registered successfully with ID: $dasId")
+    try {
+      val dasId = dasSdkManager.registerDAS(
+        request.getDefinition.getType,
+        request.getDefinition.getOptionsMap.asScala.toMap,
+        maybeDasId = if (request.hasId) Some(request.getId) else None)
+      responseObserver.onNext(dasId)
+      responseObserver.onCompleted()
+      logger.debug(s"DAS registered successfully with ID: $dasId")
+    } catch {
+      case ex: DASSdkInvalidArgumentException =>
+        logger.error("DASSdk invalid argument error", ex)
+        responseObserver.onError(Status.INVALID_ARGUMENT.withDescription(ex.getMessage).asRuntimeException())
+      case ex: DASSdkPermissionDeniedException =>
+        logger.error("DASSdk permission denied error", ex)
+        responseObserver.onError(Status.PERMISSION_DENIED.withDescription(ex.getMessage).asRuntimeException())
+      case ex: DASSdkUnauthenticatedException =>
+        logger.error("DASSdk unauthenticated error", ex)
+        responseObserver.onError(Status.UNAUTHENTICATED.withDescription(ex.getMessage).asRuntimeException())
+      case ex: DASSdkUnsupportedException =>
+        logger.error("DASSdk unsupported feature", ex)
+        responseObserver.onError(Status.UNIMPLEMENTED.withDescription(ex.getMessage).asRuntimeException())
+      case t: Throwable =>
+        logger.error("DASSdk unexpected error", t)
+        responseObserver.onError(Status.INTERNAL.withCause(t).asRuntimeException())
+
+    }
   }
 
   /**

src/main/scala/com/rawlabs/das/server/grpc/TableServiceGrpcImpl.scala

@@ -425,10 +425,15 @@ class TableServiceGrpcImpl(
           case ex: DASSdkInvalidArgumentException =>
             logger.error("DASSdk invalid argument error", ex)
             responseObserver.onError(Status.INVALID_ARGUMENT.withDescription(ex.getMessage).asRuntimeException())
+          case ex: DASSdkPermissionDeniedException =>
+            logger.error("DASSdk permission denied error", ex)
+            responseObserver.onError(Status.PERMISSION_DENIED.withDescription(ex.getMessage).asRuntimeException())
+          case ex: DASSdkUnauthenticatedException =>
+            logger.error("DASSdk unauthenticated error", ex)
+            responseObserver.onError(Status.UNAUTHENTICATED.withDescription(ex.getMessage).asRuntimeException())
           case ex: DASSdkUnsupportedException =>
             logger.error("DASSdk unsupported feature", ex)
-            responseObserver.onError(
-              Status.UNIMPLEMENTED.withDescription("Unsupported operation").withCause(ex).asRuntimeException())
+            responseObserver.onError(Status.UNIMPLEMENTED.withDescription(ex.getMessage).asRuntimeException())
           case t: Throwable =>
             logger.error("DASSdk unexpected error", t)
             responseObserver.onError(Status.INTERNAL.withCause(t).asRuntimeException())

src/main/scala/com/rawlabs/das/server/manager/DASSdkManager.scala

@@ -19,12 +19,9 @@ import scala.jdk.CollectionConverters._
 import scala.jdk.OptionConverters._
 import scala.util.control.NonFatal
 
-import com.google.common.cache.CacheBuilder
-import com.google.common.cache.CacheLoader
-import com.google.common.cache.RemovalNotification
-import com.rawlabs.das.sdk.DASSdk
-import com.rawlabs.das.sdk.DASSdkBuilder
-import com.rawlabs.das.sdk.DASSettings
+import com.google.common.cache.{CacheBuilder, CacheLoader, RemovalNotification}
+import com.google.common.util.concurrent.UncheckedExecutionException
+import com.rawlabs.das.sdk._
 import com.rawlabs.protocol.das.v1.common.DASId
 import com.rawlabs.protocol.das.v1.services.RegisterResponse
 import com.typesafe.scalalogging.StrictLogging
@@ -103,13 +100,15 @@ class DASSdkManager(implicit settings: DASSettings) extends StrictLogging {
       dasSdkCache.get(config) // If the config didn't exist, that blocks until the new DAS is loaded
       RegisterResponse.newBuilder().setId(dasId).build()
     } catch {
-      case NonFatal(e) =>
-        logger.error(s"Failed to create DAS for type: $dasType with id: $dasId", e)
+      case e: UncheckedExecutionException =>
+        // `dasSdkCache.get` throws that exception when an unchecked exception occurs while loading
+        // a missing key. Strip the Guava wrapping and rethrow the original exception.
+        logger.error(s"Failed to create DAS for type: $dasType with id: $dasId", e.getCause)
         // Remove the broken config since we failed to build the DAS
         dasSdkConfigCacheLock.synchronized {
           dasSdkConfigCache.remove(dasId)
         }
-        RegisterResponse.newBuilder().setError(e.getMessage).build()
+        throw e.getCause
     }
   }