diff --git a/common/utils/src/main/scala/org/apache/spark/internal/Logging.scala b/common/utils/src/main/scala/org/apache/spark/internal/Logging.scala
index 84b9debb2afd..a4c7d0a46e27 100644
--- a/common/utils/src/main/scala/org/apache/spark/internal/Logging.scala
+++ b/common/utils/src/main/scala/org/apache/spark/internal/Logging.scala
@@ -49,6 +49,19 @@ case class MessageWithContext(message: String, context: java.util.HashMap[String
     resultMap.putAll(mdc.context)
     MessageWithContext(message + mdc.message, resultMap)
   }
+
+  def ++(s: String): MessageWithContext = {
+    MessageWithContext(message + s, context)
+  }
+}
+
+object MDC {
+
+  implicit class StringImprovements(val s: String) {
+    def ++(mdc: MessageWithContext): MessageWithContext = {
+     MessageWithContext(s + mdc.message, mdc.context)
+    }
+  }
 }
 
 /**
@@ -107,7 +120,6 @@ trait Logging {
         if (Logging.isStructuredLoggingEnabled) {
           context.put(mdc.key.toString.toLowerCase(Locale.ROOT), mdc.value.toString)
         }
-
         if (processedParts.hasNext) {
           sb.append(processedParts.next())
         }
diff --git a/common/utils/src/test/scala/org/apache/spark/util/PatternLoggingSuite.scala b/common/utils/src/test/scala/org/apache/spark/util/PatternLoggingSuite.scala
index a75e01161d27..4d2241b05ed5 100644
--- a/common/utils/src/test/scala/org/apache/spark/util/PatternLoggingSuite.scala
+++ b/common/utils/src/test/scala/org/apache/spark/util/PatternLoggingSuite.scala
@@ -44,6 +44,14 @@ class PatternLoggingSuite extends LoggingSuiteBase with BeforeAndAfterAll {
   override def expectedPatternForMsgWithMDCAndException(level: Level): String =
     s""".*$level $className: Error in executor 1.\njava.lang.RuntimeException: OOM\n[\\s\\S]*"""
 
+  override def expectedPatternForConcatStringAndMDC(level: Level): String = {
+    s""".*$level $className: Hello This is a log message, Lost executor 1.\n"""
+  }
+
+  override def expectedPatternForConcatMDCAndString(level: Level): String = {
+    s""".*$level $className: Lost executor 1. Hello This is a log message.\n"""
+  }
+
   override def verifyMsgWithConcat(level: Level, logOutput: String): Unit = {
     val pattern =
       s""".*$level $className: Min Size: 2, Max Size: 4. Please double check.\n"""
diff --git a/common/utils/src/test/scala/org/apache/spark/util/StructuredLoggingSuite.scala b/common/utils/src/test/scala/org/apache/spark/util/StructuredLoggingSuite.scala
index 6bdd932561b5..708a23f0cca1 100644
--- a/common/utils/src/test/scala/org/apache/spark/util/StructuredLoggingSuite.scala
+++ b/common/utils/src/test/scala/org/apache/spark/util/StructuredLoggingSuite.scala
@@ -27,6 +27,7 @@ import org.scalatest.funsuite.AnyFunSuite // scalastyle:ignore funsuite
 
 import org.apache.spark.internal.{LogEntry, Logging, MDC}
 import org.apache.spark.internal.LogKey.{EXECUTOR_ID, MAX_SIZE, MIN_SIZE}
+import org.apache.spark.internal.MDC._
 
 trait LoggingSuiteBase
     extends AnyFunSuite // scalastyle:ignore funsuite
@@ -62,6 +63,12 @@ trait LoggingSuiteBase
     log"Max Size: ${MDC(MAX_SIZE, "4")}. " +
     log"Please double check."
 
+  def concatMDCAndString: LogEntry = log"Lost executor ${MDC(EXECUTOR_ID, "1")}." ++
+    s" Hello $basicMsg."
+
+  def concatStringAndMDC: LogEntry = s"Hello $basicMsg, " ++
+    log"Lost executor ${MDC(EXECUTOR_ID, "1")}."
+
   // test for basic message (without any mdc)
   def expectedPatternForBasicMsg(level: Level): String
 
@@ -74,6 +81,13 @@ trait LoggingSuiteBase
   // test for message and exception
   def expectedPatternForMsgWithMDCAndException(level: Level): String
 
+  // test for concat string and MDC
+  def expectedPatternForConcatStringAndMDC(level: Level): String
+
+  // test for concat MDC and string
+  def expectedPatternForConcatMDCAndString(level: Level): String
+
+  // test for concat MDC and MDC
   def verifyMsgWithConcat(level: Level, logOutput: String): Unit
 
   test("Basic logging") {
@@ -130,6 +144,28 @@ trait LoggingSuiteBase
           verifyMsgWithConcat(level, logOutput)
       }
   }
+
+  test("Logging concat string and MDC") {
+    Seq(
+      (Level.ERROR, () => logError(concatStringAndMDC)),
+      (Level.WARN, () => logWarning(concatStringAndMDC)),
+      (Level.INFO, () => logInfo(concatStringAndMDC))).foreach {
+        case (level, logFunc) =>
+          val logOutput = captureLogOutput(logFunc)
+          assert(expectedPatternForConcatStringAndMDC(level).r.matches(logOutput))
+      }
+  }
+
+  test("Logging concat MDC and string") {
+    Seq(
+      (Level.ERROR, () => logError(concatMDCAndString)),
+      (Level.WARN, () => logWarning(concatMDCAndString)),
+      (Level.INFO, () => logInfo(concatMDCAndString))).foreach {
+        case (level, logFunc) =>
+          val logOutput = captureLogOutput(logFunc)
+          assert(expectedPatternForConcatMDCAndString(level).r.matches(logOutput))
+      }
+  }
 }
 
 class StructuredLoggingSuite extends LoggingSuiteBase {
@@ -204,6 +240,34 @@ class StructuredLoggingSuite extends LoggingSuiteBase {
         }""")
   }
 
+  override def expectedPatternForConcatStringAndMDC(level: Level): String = {
+    compactAndToRegexPattern(
+      s"""
+        {
+          "ts": "<timestamp>",
+          "level": "$level",
+          "msg": "Hello This is a log message, Lost executor 1.",
+          "context": {
+             "executor_id": "1"
+          },
+          "logger": "$className"
+        }""")
+  }
+
+  override def expectedPatternForConcatMDCAndString(level: Level): String = {
+    compactAndToRegexPattern(
+      s"""
+        {
+          "ts": "<timestamp>",
+          "level": "$level",
+          "msg": "Lost executor 1. Hello This is a log message.",
+          "context": {
+             "executor_id": "1"
+          },
+          "logger": "$className"
+        }""")
+  }
+
   override def verifyMsgWithConcat(level: Level, logOutput: String): Unit = {
     val pattern1 = compactAndToRegexPattern(
       s"""