diff --git a/core/src/main/java/org/apache/spark/SparkThrowable.java b/core/src/main/java/org/apache/spark/SparkThrowable.java
index 52fd64135a95c..0fe3b51d3772f 100644
--- a/core/src/main/java/org/apache/spark/SparkThrowable.java
+++ b/core/src/main/java/org/apache/spark/SparkThrowable.java
@@ -19,6 +19,9 @@
 
 import org.apache.spark.annotation.Evolving;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
  * Interface mixed into Throwables thrown from Spark.
  *
@@ -51,13 +54,8 @@ default boolean isInternalError() {
     return SparkThrowableHelper.isInternalError(this.getErrorClass());
   }
 
-  default String[] getMessageParameters() {
-    return new String[]{};
-  }
-
-  // Returns a string array of all parameters that need to be passed to this error message.
-  default String[] getParameterNames() {
-    return SparkThrowableHelper.getParameterNames(this.getErrorClass(), this.getErrorSubClass());
+  default Map<String, String> getMessageParameters() {
+    return new HashMap();
   }
 
   default QueryContext[] getQueryContext() { return new QueryContext[0]; }
diff --git a/core/src/main/java/org/apache/spark/memory/SparkOutOfMemoryError.java b/core/src/main/java/org/apache/spark/memory/SparkOutOfMemoryError.java
index cfbf2e574787d..0916d9e6fe307 100644
--- a/core/src/main/java/org/apache/spark/memory/SparkOutOfMemoryError.java
+++ b/core/src/main/java/org/apache/spark/memory/SparkOutOfMemoryError.java
@@ -47,8 +47,8 @@ public SparkOutOfMemoryError(String errorClass, Map<String, String> messageParam
     }
 
     @Override
-    public String[] getMessageParameters() {
-        return SparkThrowableHelper.getMessageParameters(errorClass, null, messageParameters);
+    public Map<String, String> getMessageParameters() {
+        return messageParameters;
     }
 
     @Override
diff --git a/core/src/main/scala/org/apache/spark/SparkException.scala b/core/src/main/scala/org/apache/spark/SparkException.scala
index dba6ef9347ff0..ff59cfb3455f1 100644
--- a/core/src/main/scala/org/apache/spark/SparkException.scala
+++ b/core/src/main/scala/org/apache/spark/SparkException.scala
@@ -22,6 +22,8 @@ import java.sql.{SQLException, SQLFeatureNotSupportedException}
 import java.time.DateTimeException
 import java.util.ConcurrentModificationException
 
+import scala.collection.JavaConverters._
+
 import org.apache.hadoop.fs.FileAlreadyExistsException
 
 class SparkException(
@@ -86,11 +88,7 @@ class SparkException(
       errorSubClass = Some(errorSubClass),
       messageParameters = messageParameters)
 
-  override def getMessageParameters: Array[String] = {
-    errorClass.map { ec =>
-      SparkThrowableHelper.getMessageParameters(ec, errorSubClass.orNull, messageParameters)
-    }.getOrElse(Array.empty)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass.orNull
   override def getErrorSubClass: String = errorSubClass.orNull
@@ -146,9 +144,7 @@ private[spark] class SparkUpgradeException(
     SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters), cause)
   with SparkThrowable {
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull}
@@ -166,9 +162,7 @@ private[spark] class SparkArithmeticException(
     SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, summary))
   with SparkThrowable {
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
@@ -195,9 +189,7 @@ private[spark] class SparkUnsupportedOperationException(
       errorSubClass = Some(errorSubClass),
       messageParameters = messageParameters)
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
@@ -215,9 +207,7 @@ private[spark] class SparkClassNotFoundException(
     SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters), cause)
   with SparkThrowable {
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull}
@@ -234,9 +224,7 @@ private[spark] class SparkConcurrentModificationException(
     SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters), cause)
   with SparkThrowable {
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull}
@@ -254,9 +242,7 @@ private[spark] class SparkDateTimeException(
     SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, summary))
   with SparkThrowable {
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
@@ -274,9 +260,7 @@ private[spark] class SparkFileAlreadyExistsException(
     SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
   with SparkThrowable {
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull}
@@ -292,9 +276,7 @@ private[spark] class SparkFileNotFoundException(
     SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
   with SparkThrowable {
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull}
@@ -312,9 +294,7 @@ private[spark] class SparkNumberFormatException(
     SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, summary))
   with SparkThrowable {
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
@@ -334,9 +314,7 @@ private[spark] class SparkIllegalArgumentException(
     SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, summary))
   with SparkThrowable {
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
@@ -379,9 +357,7 @@ private[spark] class SparkRuntimeException(
       cause = null,
       context = Array.empty[QueryContext])
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
@@ -399,9 +375,7 @@ private[spark] class SparkSecurityException(
     SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
   with SparkThrowable {
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
@@ -420,9 +394,7 @@ private[spark] class SparkArrayIndexOutOfBoundsException(
     SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters, summary))
   with SparkThrowable {
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
@@ -446,9 +418,7 @@ private[spark] class SparkSQLException(
       errorSubClass = None,
       messageParameters = messageParameters)
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
@@ -474,9 +444,7 @@ private[spark] class SparkSQLFeatureNotSupportedException(
       errorSubClass = Some(errorSubClass),
       messageParameters = messageParameters)
 
-  override def getMessageParameters: Array[String] = {
-    SparkThrowableHelper.getMessageParameters(errorClass, errorSubClass.orNull, messageParameters)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass
   override def getErrorSubClass: String = errorSubClass.orNull
diff --git a/core/src/main/scala/org/apache/spark/SparkThrowableHelper.scala b/core/src/main/scala/org/apache/spark/SparkThrowableHelper.scala
index 81c5c6cb043b3..497fd91d77ac7 100644
--- a/core/src/main/scala/org/apache/spark/SparkThrowableHelper.scala
+++ b/core/src/main/scala/org/apache/spark/SparkThrowableHelper.scala
@@ -101,20 +101,6 @@ private[spark] object SparkThrowableHelper {
     parameterNames
   }
 
-  def getMessageParameters(
-      errorClass: String,
-      errorSubCLass: String,
-      params: Map[String, String]): Array[String] = {
-    getParameterNames(errorClass, errorSubCLass).map(params.getOrElse(_, "?"))
-  }
-
-  def getMessageParameters(
-      errorClass: String,
-      errorSubCLass: String,
-      params: java.util.Map[String, String]): Array[String] = {
-    getParameterNames(errorClass, errorSubCLass).map(params.getOrDefault(_, "?"))
-  }
-
   def getMessage(
       errorClass: String,
       errorSubClass: String,
@@ -185,8 +171,6 @@ private[spark] object SparkThrowableHelper {
         }
       case MINIMAL | STANDARD =>
         val errorClass = e.getErrorClass
-        assert(e.getParameterNames.size == e.getMessageParameters.size,
-          "Number of message parameter names and values must be the same")
         toJsonString { generator =>
           val g = generator.useDefaultPrettyPrinter()
           g.writeStartObject()
@@ -200,10 +184,10 @@ private[spark] object SparkThrowableHelper {
           }
           val sqlState = e.getSqlState
           if (sqlState != null) g.writeStringField("sqlState", sqlState)
-          val parameterNames = e.getParameterNames
-          if (!parameterNames.isEmpty) {
+          val messageParameters = e.getMessageParameters
+          if (!messageParameters.isEmpty) {
             g.writeObjectFieldStart("messageParameters")
-            (parameterNames zip e.getMessageParameters).sortBy(_._1).foreach { case (name, value) =>
+            messageParameters.asScala.toSeq.sortBy(_._1).foreach { case (name, value) =>
               g.writeStringField(name, value)
             }
             g.writeEndObject()
diff --git a/core/src/test/scala/org/apache/spark/SparkFunSuite.scala b/core/src/test/scala/org/apache/spark/SparkFunSuite.scala
index ade20dbff83ff..10ebbe76d6c74 100644
--- a/core/src/test/scala/org/apache/spark/SparkFunSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkFunSuite.scala
@@ -23,6 +23,7 @@ import java.nio.file.{Files, Path}
 import java.util.{Locale, TimeZone}
 
 import scala.annotation.tailrec
+import scala.collection.JavaConverters._
 import scala.collection.mutable.ArrayBuffer
 
 import org.apache.commons.io.FileUtils
@@ -305,7 +306,7 @@ abstract class SparkFunSuite
       assert(exception.getErrorSubClass === errorSubClass.get)
     }
     sqlState.foreach(state => assert(exception.getSqlState === state))
-    val expectedParameters = (exception.getParameterNames zip exception.getMessageParameters).toMap
+    val expectedParameters = exception.getMessageParameters.asScala
     if (matchPVals == true) {
       assert(expectedParameters.size === parameters.size)
       expectedParameters.foreach(
diff --git a/core/src/test/scala/org/apache/spark/SparkThrowableSuite.scala b/core/src/test/scala/org/apache/spark/SparkThrowableSuite.scala
index cbf273dc5c857..6eb66cd8f18fd 100644
--- a/core/src/test/scala/org/apache/spark/SparkThrowableSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkThrowableSuite.scala
@@ -161,7 +161,7 @@ class SparkThrowableSuite extends SparkFunSuite {
       getMessage("UNRESOLVED_COLUMN", "WITHOUT_SUGGESTION", Map.empty[String, String])
     }
     assert(e.getErrorClass === "INTERNAL_ERROR")
-    assert(e.getMessageParameters.head.contains("Undefined an error message parameter"))
+    assert(e.getMessageParameters().get("message").contains("Undefined an error message parameter"))
 
     // Does not fail with too many args (expects 0 args)
     assert(getMessage("DIVIDE_BY_ZERO", null, Map("config" -> "foo", "a" -> "bar")) ==
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/AnalysisException.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/AnalysisException.scala
index 4b27dfc00c2aa..9dbdaa3d59a2b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/AnalysisException.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/AnalysisException.scala
@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql
 
+import scala.collection.JavaConverters._
+
 import org.apache.spark.{QueryContext, SparkThrowable, SparkThrowableHelper}
 import org.apache.spark.annotation.Stable
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
@@ -165,11 +167,7 @@ class AnalysisException protected[sql] (
     message
   }
 
-  override def getMessageParameters: Array[String] = {
-    errorClass.map { ec =>
-      SparkThrowableHelper.getMessageParameters(ec, errorSubClass.orNull, messageParameters)
-    }.getOrElse(Array.empty)
-  }
+  override def getMessageParameters: java.util.Map[String, String] = messageParameters.asJava
 
   override def getErrorClass: String = errorClass.orNull
   override def getErrorSubClass: String = errorSubClass.orNull