diff --git a/akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/EventSourcedBehaviorLoggingSpec.scala b/akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/EventSourcedBehaviorLoggingSpec.scala
index b4e592543c..0e0bf87163 100644
--- a/akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/EventSourcedBehaviorLoggingSpec.scala
+++ b/akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/EventSourcedBehaviorLoggingSpec.scala
@@ -15,6 +15,7 @@ import akka.persistence.typed.EventSourcedBehaviorLoggingSpec.ChattyEventSourcin
 import akka.persistence.typed.scaladsl.Effect
 import akka.persistence.typed.scaladsl.EventSourcedBehavior
 import akka.serialization.jackson.CborSerializable
+import com.typesafe.config.{ Config, ConfigFactory }
 import org.scalatest.wordspec.AnyWordSpecLike
 
 object EventSourcedBehaviorLoggingSpec {
@@ -47,17 +48,20 @@ object EventSourcedBehaviorLoggingSpec {
   }
 }
 
-class EventSourcedBehaviorLoggingSpec
-    extends ScalaTestWithActorTestKit(PersistenceTestKitPlugin.config)
+abstract class EventSourcedBehaviorLoggingSpec(config: Config)
+    extends ScalaTestWithActorTestKit(config)
     with AnyWordSpecLike
     with LogCapturing {
   import EventSourcedBehaviorLoggingSpec._
 
-  "Chatty behavior" must {
+  def loggerName: String
+  def loggerId: String
+
+  s"Chatty behavior ($loggerId)" must {
     val myId = PersistenceId("Chatty", "chat-1")
     val chattyActor = spawn(ChattyEventSourcingBehavior(myId))
 
-    "log user message in context.log" in {
+    "always log user message in context.log" in {
       LoggingTestKit
         .info("received message 'Mary'")
         .withLoggerName("akka.persistence.typed.EventSourcedBehaviorLoggingSpec$ChattyEventSourcingBehavior$")
@@ -66,32 +70,32 @@ class EventSourcedBehaviorLoggingSpec
         }
     }
 
-    "log internal messages in 'Internal' logger without logging user data (Persist)" in {
+    s"log internal messages in '$loggerId' logger without logging user data (Persist)" in {
       LoggingTestKit
         .debug(
           "Handled command [akka.persistence.typed.EventSourcedBehaviorLoggingSpec$ChattyEventSourcingBehavior$Hello], " +
           "resulting effect: [Persist(akka.persistence.typed.EventSourcedBehaviorLoggingSpec$ChattyEventSourcingBehavior$Event)], side effects: [0]")
-        .withLoggerName("akka.persistence.typed.internal.EventSourcedBehaviorImpl")
+        .withLoggerName(loggerName)
         .expect {
           chattyActor ! Hello("Joe")
         }
     }
 
-    "log internal messages in 'Internal' logger without logging user data (PersistAll)" in {
+    s"log internal messages in '$loggerId' logger without logging user data (PersistAll)" in {
       LoggingTestKit
         .debug("Handled command [akka.persistence.typed.EventSourcedBehaviorLoggingSpec$ChattyEventSourcingBehavior$Hellos], " +
         "resulting effect: [PersistAll(akka.persistence.typed.EventSourcedBehaviorLoggingSpec$ChattyEventSourcingBehavior$Event," +
         "akka.persistence.typed.EventSourcedBehaviorLoggingSpec$ChattyEventSourcingBehavior$Event)], side effects: [0]")
-        .withLoggerName("akka.persistence.typed.internal.EventSourcedBehaviorImpl")
+        .withLoggerName(loggerName)
         .expect {
           chattyActor ! Hellos("Mary", "Joe")
         }
     }
 
-    "Internal logger preserves MDC source" in {
+    s"log in '$loggerId' while preserving MDC source" in {
       LoggingTestKit
         .debug("Handled command ")
-        .withLoggerName("akka.persistence.typed.internal.EventSourcedBehaviorImpl")
+        .withLoggerName(loggerName)
         .withMdc(Map("persistencePhase" -> "running-cmd", "persistenceId" -> "Chatty|chat-1"))
         .expect {
           chattyActor ! Hello("Mary")
@@ -99,3 +103,21 @@ class EventSourcedBehaviorLoggingSpec
     }
   }
 }
+
+class EventSourcedBehaviorLoggingInternalLoggerSpec
+    extends EventSourcedBehaviorLoggingSpec(PersistenceTestKitPlugin.config) {
+  override def loggerName = "akka.persistence.typed.internal.EventSourcedBehaviorImpl"
+  override def loggerId = "internal.log"
+}
+
+object EventSourcedBehaviorLoggingContextLoggerSpec {
+  val config =
+    ConfigFactory
+      .parseString("akka.persistence.typed.use-context-logger-for-internal-logging = true")
+      .withFallback(PersistenceTestKitPlugin.config)
+}
+class EventSourcedBehaviorLoggingContextLoggerSpec
+    extends EventSourcedBehaviorLoggingSpec(EventSourcedBehaviorLoggingContextLoggerSpec.config) {
+  override def loggerName = "akka.persistence.typed.EventSourcedBehaviorLoggingSpec$ChattyEventSourcingBehavior$"
+  override def loggerId = "context.log"
+}
diff --git a/akka-persistence-typed/src/main/mima-filters/2.6.14.backwards.excludes/settings-to-fallback-to-context-log.excludes b/akka-persistence-typed/src/main/mima-filters/2.6.14.backwards.excludes/settings-to-fallback-to-context-log.excludes
new file mode 100644
index 0000000000..89351529cc
--- /dev/null
+++ b/akka-persistence-typed/src/main/mima-filters/2.6.14.backwards.excludes/settings-to-fallback-to-context-log.excludes
@@ -0,0 +1,5 @@
+# settings to fallback to actor's context.log for internal EventSourcedBehavior logging
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.persistence.typed.internal.EventSourcedSettings.apply")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("akka.persistence.typed.internal.EventSourcedSettings.unapply")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.persistence.typed.internal.EventSourcedSettings.copy")
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.persistence.typed.internal.EventSourcedSettings.this")
diff --git a/akka-persistence-typed/src/main/resources/reference.conf b/akka-persistence-typed/src/main/resources/reference.conf
index afc0976127..43500138fd 100644
--- a/akka-persistence-typed/src/main/resources/reference.conf
+++ b/akka-persistence-typed/src/main/resources/reference.conf
@@ -39,6 +39,12 @@ akka.persistence.typed {
   # enables automatic DEBUG level logging of messages stashed automatically by an EventSourcedBehavior,
   # this may happen while it receives commands while it is recovering events or while it is persisting events
   log-stashing = off
+
+  # By default, internal event sourced behavior logging are sent to
+  # akka.persistence.typed.internal.EventSourcedBehaviorImpl
+  # this can be changed by setting this to 'true' in which case the internal logging is sent to
+  # the actor context logger.
+  use-context-logger-for-internal-logging = false
 }
 
 akka.reliable-delivery {
diff --git a/akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventSourcedBehaviorImpl.scala b/akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventSourcedBehaviorImpl.scala
index 335e125368..878028f6a7 100644
--- a/akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventSourcedBehaviorImpl.scala
+++ b/akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventSourcedBehaviorImpl.scala
@@ -108,7 +108,7 @@ private[akka] final case class EventSourcedBehaviorImpl[Command, Event, State](
     throw new IllegalArgumentException("persistenceId must not be null")
 
   // Don't use it directly, but instead call internalLogger() (see below)
-  private val logger = LoggerFactory.getLogger(this.getClass)
+  private val loggerForInternal = LoggerFactory.getLogger(this.getClass)
 
   override def apply(context: typed.TypedActorContext[Command]): Behavior[Command] = {
     val ctx = context.asScala
@@ -124,10 +124,13 @@ private[akka] final case class EventSourcedBehaviorImpl[Command, Event, State](
 
     // This method ensures that the MDC is set before we use the internal logger
     def internalLogger() = {
-      // MDC is cleared (if used) from aroundReceive in ActorAdapter after processing each message,
-      // but important to call `context.log` to mark MDC as used
-      ctx.log
-      logger
+      if (settings.useContextLoggerForInternalLogging) ctx.log
+      else {
+        // MDC is cleared (if used) from aroundReceive in ActorAdapter after processing each message,
+        // but important to call `context.log` to mark MDC as used
+        ctx.log
+        loggerForInternal
+      }
     }
 
     val actualSignalHandler: PartialFunction[(State, Signal), Unit] = signalHandler.orElse {
diff --git a/akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventSourcedSettings.scala b/akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventSourcedSettings.scala
index 50aeb57189..e6716137b3 100644
--- a/akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventSourcedSettings.scala
+++ b/akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventSourcedSettings.scala
@@ -41,6 +41,8 @@ import akka.persistence.Persistence
     val recoveryEventTimeout: FiniteDuration =
       journalConfig.getDuration("recovery-event-timeout", TimeUnit.MILLISECONDS).millis
 
+    val useContextLoggerForInternalLogging = typedConfig.getBoolean("use-context-logger-for-internal-logging")
+
     Persistence.verifyPluginConfigExists(config, snapshotPluginId, "Snapshot store")
 
     EventSourcedSettings(
@@ -49,7 +51,8 @@ import akka.persistence.Persistence
       logOnStashing = logOnStashing,
       recoveryEventTimeout,
       journalPluginId,
-      snapshotPluginId)
+      snapshotPluginId,
+      useContextLoggerForInternalLogging)
   }
 
   private def journalConfigFor(config: Config, journalPluginId: String): Config = {
@@ -76,7 +79,8 @@ private[akka] final case class EventSourcedSettings(
     logOnStashing: Boolean,
     recoveryEventTimeout: FiniteDuration,
     journalPluginId: String,
-    snapshotPluginId: String) {
+    snapshotPluginId: String,
+    useContextLoggerForInternalLogging: Boolean) {
 
   require(journalPluginId != null, "journal plugin id must not be null; use empty string for 'default' journal")
   require(
diff --git a/akka-persistence-typed/src/test/scala/akka/persistence/typed/internal/StashStateSpec.scala b/akka-persistence-typed/src/test/scala/akka/persistence/typed/internal/StashStateSpec.scala
index 7ef468b403..f87f22ce97 100644
--- a/akka-persistence-typed/src/test/scala/akka/persistence/typed/internal/StashStateSpec.scala
+++ b/akka-persistence-typed/src/test/scala/akka/persistence/typed/internal/StashStateSpec.scala
@@ -66,6 +66,7 @@ class StashStateSpec extends ScalaTestWithActorTestKit with AnyWordSpecLike with
       logOnStashing = false,
       recoveryEventTimeout = 3.seconds,
       journalPluginId = "",
-      snapshotPluginId = "")
+      snapshotPluginId = "",
+      useContextLoggerForInternalLogging = false)
 
 }