diff --git a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java
index 0c6904791d8..91d71819721 100644
--- a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java
+++ b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java
@@ -47,6 +47,7 @@ import org.skife.jdbi.v2.StatementContext;
 import org.skife.jdbi.v2.tweak.HandleCallback;
 import org.skife.jdbi.v2.tweak.ResultSetMapper;
 
+import javax.annotation.Nullable;
 import java.io.IOException;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -244,6 +245,7 @@ public class SQLMetadataSupervisorManager implements MetadataSupervisorManager
                 ).map(
                     new ResultSetMapper<Pair<String, SupervisorSpec>>()
                     {
+                      @Nullable
                       @Override
                       public Pair<String, SupervisorSpec> map(int index, ResultSet r, StatementContext ctx)
                           throws SQLException
@@ -259,7 +261,13 @@ public class SQLMetadataSupervisorManager implements MetadataSupervisorManager
                           );
                         }
                         catch (IOException e) {
-                          throw new RuntimeException(e);
+                          String exceptionMessage = StringUtils.format(
+                              "Could not map json payload to a SupervisorSpec for spec_id: [%s]."
+                              + " Delete the supervisor from the database and re-submit it to the overlord.",
+                              r.getString("spec_id")
+                          );
+                          log.error(e, exceptionMessage);
+                          return null;
                         }
                       }
                     }
@@ -276,7 +284,9 @@ public class SQLMetadataSupervisorManager implements MetadataSupervisorManager
                       )
                       {
                         try {
-                          retVal.put(stringObjectMap.lhs, stringObjectMap.rhs);
+                          if (null != stringObjectMap) {
+                            retVal.put(stringObjectMap.lhs, stringObjectMap.rhs);
+                          }
                           return retVal;
                         }
                         catch (Exception e) {
diff --git a/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java b/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java
index 7ea26960637..b43958fc746 100644
--- a/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java
+++ b/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java
@@ -275,6 +275,9 @@ public class SQLMetadataSupervisorManagerTest
     specs = allSpecs.get(supervisor2);
     Assert.assertEquals(1, specs.size());
     Assert.assertNull(specs.get(0).getSpec());
+
+    Map<String, SupervisorSpec> latestSupervisorSpec = supervisorManager.getLatest();
+    Assert.assertEquals(1, latestSupervisorSpec.size());
   }
 
   @Test