diff --git a/cdap-api/src/main/java/io/cdap/cdap/api/service/ServiceException.java b/cdap-api/src/main/java/io/cdap/cdap/api/service/ServiceException.java
new file mode 100644
index 000000000000..08f40b5bd722
--- /dev/null
+++ b/cdap-api/src/main/java/io/cdap/cdap/api/service/ServiceException.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright © 2024 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.cdap.api.service;
+
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+
+/**
+ * Similar to {@link ServiceUnavailableException}, but responds with a custom error code.
+ */
+public class ServiceException extends ServiceUnavailableException {
+
+  public static final List<Integer> HTTP_SERVER_ERROR_CODES = Collections.unmodifiableList(
+      Arrays.asList(HttpURLConnection.HTTP_BAD_GATEWAY,
+          HttpURLConnection.HTTP_GATEWAY_TIMEOUT,
+          HttpURLConnection.HTTP_INTERNAL_ERROR,
+          HttpURLConnection.HTTP_UNAVAILABLE));
+  private final int statusCode;
+  private final String jsonDetails;
+
+  /**
+   * Constructs {@link ServiceException}.
+   *
+   * @param serviceName the canonical name of service
+   * @param message the message
+   * @param jsonDetails the json details
+   * @param cause the cause {@link Throwable}
+   * @param statusCode the status code
+   */
+  public ServiceException(String serviceName, String message, @Nullable String jsonDetails,
+      @Nullable Throwable cause, int statusCode) throws IOException {
+    super(serviceName, message, cause);
+
+    if (!HTTP_SERVER_ERROR_CODES.contains(statusCode)) {
+      throw new IOException(
+          "Cannot instantiate Service Exception, it is not a retryable response code");
+    }
+
+    this.statusCode = statusCode;
+    this.jsonDetails = jsonDetails;
+  }
+
+  @Nullable
+  public String getJsonDetails() {
+    return jsonDetails;
+  }
+
+  @Override
+  public int getStatusCode() {
+    return statusCode;
+  }
+}
diff --git a/cdap-api/src/main/java/io/cdap/cdap/api/service/ServiceUnavailableException.java b/cdap-api/src/main/java/io/cdap/cdap/api/service/ServiceUnavailableException.java
index 52708b217c88..6151085d435f 100644
--- a/cdap-api/src/main/java/io/cdap/cdap/api/service/ServiceUnavailableException.java
+++ b/cdap-api/src/main/java/io/cdap/cdap/api/service/ServiceUnavailableException.java
@@ -19,6 +19,7 @@
 import io.cdap.cdap.api.common.HttpErrorStatusProvider;
 import io.cdap.cdap.api.retry.RetryableException;
 import java.net.HttpURLConnection;
+import javax.annotation.Nullable;
 
 /**
  * Exception thrown when the service is not running.
@@ -26,11 +27,12 @@
 public class ServiceUnavailableException extends RetryableException implements
     HttpErrorStatusProvider {
 
-  private final String serviceName;
+  protected static final String MESSAGE_FORMAT =
+      "Service %s is not available. Please wait until it is up and running.";
+  protected final String serviceName;
 
   public ServiceUnavailableException(String serviceName) {
-    this(serviceName,
-        "Service '" + serviceName + "' is not available. Please wait until it is up and running.");
+    this(serviceName, String.format(MESSAGE_FORMAT, serviceName));
   }
 
   public ServiceUnavailableException(String serviceName, String message) {
@@ -39,9 +41,7 @@ public ServiceUnavailableException(String serviceName, String message) {
   }
 
   public ServiceUnavailableException(String serviceName, Throwable cause) {
-    this(serviceName,
-        "Service '" + serviceName + "' is not available. Please wait until it is up and running.",
-        cause);
+    this(serviceName, String.format(MESSAGE_FORMAT, serviceName), cause);
   }
 
   public ServiceUnavailableException(String serviceName, String message, Throwable cause) {
diff --git a/cdap-common/src/main/java/io/cdap/cdap/common/internal/remote/RemoteClient.java b/cdap-common/src/main/java/io/cdap/cdap/common/internal/remote/RemoteClient.java
index 0020b8565ca0..b07af7bb5f37 100644
--- a/cdap-common/src/main/java/io/cdap/cdap/common/internal/remote/RemoteClient.java
+++ b/cdap-common/src/main/java/io/cdap/cdap/common/internal/remote/RemoteClient.java
@@ -22,12 +22,11 @@
 import com.google.common.net.HttpHeaders;
 import io.cdap.cdap.api.retry.Idempotency;
 import io.cdap.cdap.api.retry.RetryableException;
-import io.cdap.cdap.common.ServiceException;
+import io.cdap.cdap.api.service.ServiceException;
 import io.cdap.cdap.api.service.ServiceUnavailableException;
 import io.cdap.cdap.common.discovery.EndpointStrategy;
 import io.cdap.cdap.common.discovery.RandomEndpointStrategy;
 import io.cdap.cdap.common.discovery.URIScheme;
-import io.cdap.cdap.common.http.HttpCodes;
 import io.cdap.cdap.common.security.HttpsEnabler;
 import io.cdap.cdap.proto.security.Credential;
 import io.cdap.cdap.security.spi.authenticator.RemoteAuthenticator;
@@ -38,7 +37,6 @@
 import io.cdap.common.http.HttpRequestConfig;
 import io.cdap.common.http.HttpRequests;
 import io.cdap.common.http.HttpResponse;
-import io.netty.handler.codec.http.HttpResponseStatus;
 import java.io.IOException;
 import java.net.ConnectException;
 import java.net.HttpURLConnection;
@@ -168,7 +166,7 @@ private HttpResponse executeNonIdempotent(HttpRequest request) throws IOExceptio
         throw new ServiceUnavailableException(discoverableServiceName,
             response.getResponseBodyAsString());
       }
-      if (HttpCodes.isRetryable(responseCode)) {
+      if (ServiceException.HTTP_SERVER_ERROR_CODES.contains(responseCode)) {
         String contentType = response.getHeaders().get(HttpHeaders.CONTENT_TYPE).stream()
             .findFirst().orElse(null);
         String message;
@@ -181,8 +179,8 @@ private HttpResponse executeNonIdempotent(HttpRequest request) throws IOExceptio
           message = String.format("Service %s is not available: %s", discoverableServiceName,
               response.getResponseBodyAsString());
         }
-        throw new ServiceException(message, null,
-            jsonDetails, HttpResponseStatus.valueOf(responseCode));
+        throw new ServiceException(discoverableServiceName, message,
+            jsonDetails, null, responseCode);
       }
       if (responseCode == HttpURLConnection.HTTP_FORBIDDEN) {
         throw new UnauthorizedException(response.getResponseBodyAsString());
diff --git a/cdap-common/src/main/java/io/cdap/cdap/common/internal/remote/RemoteTaskExecutor.java b/cdap-common/src/main/java/io/cdap/cdap/common/internal/remote/RemoteTaskExecutor.java
index 7c00a2452906..e0a593472d3e 100644
--- a/cdap-common/src/main/java/io/cdap/cdap/common/internal/remote/RemoteTaskExecutor.java
+++ b/cdap-common/src/main/java/io/cdap/cdap/common/internal/remote/RemoteTaskExecutor.java
@@ -142,8 +142,19 @@ public byte[] runTask(RunnableTaskRequest runnableTaskRequest) throws Exception
                   runnableTaskRequest.getClassName()));
         }
       }, retryStrategy, Retries.DEFAULT_PREDICATE);
-    } catch (ServiceException se) {
-      Exception ex = getTaskException(se);
+    } catch (ServiceException | io.cdap.cdap.api.service.ServiceException se) {
+      Exception ex = se;
+      if (se instanceof ServiceException && ((ServiceException) se).getJsonDetails() != null) {
+        ex = getTaskException(((ServiceException) se).getJsonDetails());
+      } else if (se instanceof io.cdap.cdap.api.service.ServiceException
+          && ((io.cdap.cdap.api.service.ServiceException) se).getJsonDetails() != null) {
+        ex = getTaskException(((io.cdap.cdap.api.service.ServiceException) se).getJsonDetails());
+      }
+
+      if (ex instanceof JsonSyntaxException) {
+        se.addSuppressed(ex);
+        ex = se;
+      }
       //emit metrics with failed result
       emitMetrics(startTime, false, runnableTaskRequest, getAttempts(ex));
       throw ex;
@@ -154,17 +165,12 @@ public byte[] runTask(RunnableTaskRequest runnableTaskRequest) throws Exception
     }
   }
 
-  private Exception getTaskException(ServiceException e) {
-    if (e.getJsonDetails() == null) {
-      // This is not an application-level exception, might be a timeout or similar failure
-      return e;
-    }
+  private Exception getTaskException(String jsonDetails) {
     try {
-      BasicThrowable basicThrowable = GSON.fromJson(e.getJsonDetails(), BasicThrowable.class);
+      BasicThrowable basicThrowable = GSON.fromJson(jsonDetails, BasicThrowable.class);
       return RemoteExecutionException.fromBasicThrowable(basicThrowable);
     } catch (JsonSyntaxException jse) {
-      e.addSuppressed(jse);
-      return e;
+      return jse;
     }
   }