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

[bgaidioz]

• Added a couple needed new exceptions for an SDK to report specific errors,
• Catch the exceptions in the server to rewrite them as gRPC errors,
• Edited DASSdkManager to catch the guava cache exception one gets if the SDK throws an exception, so that the SDK exception propagates. Otherwise, whichever exception propagates. Therefore I removed the .setError code to be consistent with the idea that errors are reported through gRPC errors.

Caution

I believe we never hit REGISTRATION_FAILED in psql, since that one is made up from .error instead of a gRPC failure. Maybe for backward compatibility we can instead leave it?

Manually tried against that version of multicorn-das: raw-labs/multicorn-das#27 which modifies error handling during registration. With errors triggered in DASMock or DASMockTable:

If for example DASMock throws a DASSdkUnauthenticatedException in its
initializer. That mimics a DAS instance that performs some authentication early
on, verifies the password, etc.

class DASMock(options: Map[String, String]) extends DASSdk with StrictLogging {

  options.keys.foreach(key => logger.info(s"Option: $key = ${options(key)}"))

  private val dasMockStorage = new DASMockStorage("column1")
  throw new DASSdkUnauthenticatedException("Wrong password") // HERE

The effect is:

ERROR:  Wrong password
DETAIL:  {"code": "UNAUTHENTICATED", "message": "Wrong password", "das_name":
"localhost:50051", "das_type": "mock", "das_url": "localhost:50051",
"table_name": "slow", "cause": "<_InactiveRpcError of RPC that terminated
with:\n\tstatus = StatusCode.UNAUTHENTICATED\n\tdetails = \"Wrong
password\"\n\tdebug_error_string = \"UNKNOWN:Error received from peer
{created_time:\"2025-02-28T14:35:21.707372+01:00\", grpc_status:16,
grpc_message:\"Wrong password\"}\"\n>"}

Similar with an invalid option.

throw new DASSdkInvalidArgumentException("Invalid options: missing an important key")

ERROR:  Invalid options: missing an important key
DETAIL:  {"code": "INVALID_ARGUMENT", "message": "Invalid options: missing an
important key", "das_name": "localhost:50051", "das_type": "mock", "das_url":
"localhost:50051", "table_name": "slow", "cause": "<_InactiveRpcError of RPC
that terminated with:\n\tstatus = StatusCode.INVALID_ARGUMENT\n\tdetails =
\"Invalid options: missing an important key\"\n\tdebug_error_string =
\"UNKNOWN:Error received from peer
{created_time:\"2025-02-28T14:39:34.398676+01:00\", grpc_status:3,
grpc_message:\"Invalid options: missing an important key\"}\"\n>"}

If the error is unexpected (e.g. a bug in the DAS initializer, NullPointerException). The error is INTERNAL.

ERROR:  gRPC error calling remote DAS server
DETAIL:  {"code": "INTERNAL", "message": "gRPC error calling remote DAS
server", "das_name": null, "das_type": null, "das_url": null, "table_name":
null, "cause": "<_InactiveRpcError of RPC that terminated with:\n\tstatus =
StatusCode.INTERNAL\n\tdetails = \"\"\n\tdebug_error_string = \"UNKNOWN:Error
received from peer  {created_time:\"2025-02-28T14:38:26.791684+01:00\",
grpc_status:13, grpc_message:\"\"}\"\n>"}

And if the DAS server is down, after 30 attempts:

ERROR:  Server unavailable
DETAIL:  {"code": "UNAVAILABLE", "message": "Server unavailable", "das_name":
"localhost:50051", "das_type": "mock", "das_url": "localhost:50051",
"table_name": "slow", "cause": "<_InactiveRpcError of RPC that terminated
with:\n\tstatus = StatusCode.UNAVAILABLE\n\tdetails = \"failed to connect to
all addresses; last error: UNKNOWN: ipv4:127.0.0.1:50051: Failed to connect to
remote host: connect: Connection refused (61)\"\n\tdebug_error_string =
\"UNKNOWN:Error received from peer  {grpc_message:\"failed to connect to all
addresses; last error: UNKNOWN: ipv4:127.0.0.1:50051: Failed to connect to
remote host: connect: Connection refused (61)\", grpc_status:14,
created_time:\"2025-02-28T14:41:27.460325+01:00\"}\"\n>"}

If the error occurs after registration, e.g. when running .execute of a DASTable.

    throw new DASSdkInvalidArgumentException("Invalid predicate")

ERROR:  Invalid predicate
DETAIL:  {"code": "INVALID_ARGUMENT", "message": "Invalid predicate",
"das_name": "localhost:50051", "das_type": "mock", "das_url":
"localhost:50051", "table_name": "slow", "cause": "<_MultiThreadedRendezvous of
RPC that terminated with:\n\tstatus = StatusCode.INVALID_ARGUMENT\n\tdetails =
\"Invalid predicate\"\n\tdebug_error_string = \"UNKNOWN:Error received from
peer  {grpc_message:\"Invalid predicate\", grpc_status:3,
created_time:\"2025-02-28T14:43:35.637285+01:00\"}\"\n>"}

    throw new DASSdkUnauthenticatedException("Who are you?")

WARNING:  gRPC attempting registration and retry...
ERROR:  Who are you?
DETAIL:  {"code": "UNAUTHENTICATED", "message": "Who are you?", "das_name":
"localhost:50051", "das_type": "mock", "das_url": "localhost:50051",
"table_name": "slow", "cause": "<_MultiThreadedRendezvous of RPC that
terminated with:\n\tstatus = StatusCode.UNAUTHENTICATED\n\tdetails = \"Who are
you?\"\n\tdebug_error_string = \"UNKNOWN:Error received from peer
{grpc_message:\"Who are you?\", grpc_status:16,
created_time:\"2025-02-28T14:44:37.911759+01:00\"}\"\n>"}

    throw new DASSdkPermissionDeniedException("Forbidden")

ERROR:  Forbidden
DETAIL:  {"code": "PERMISSION_DENIED", "message": "Forbidden", "das_name":
"localhost:50051", "das_type": "mock", "das_url": "localhost:50051",
"table_name": "slow", "cause": "<_MultiThreadedRendezvous of RPC that
terminated with:\n\tstatus = StatusCode.PERMISSION_DENIED\n\tdetails =
\"Forbidden\"\n\tdebug_error_string = \"UNKNOWN:Error received from peer
{grpc_message:\"Forbidden\", grpc_status:7,
created_time:\"2025-02-28T14:46:08.608081+01:00\"}\"\n>"}