diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/TaskMigratedException.java b/streams/src/main/java/org/apache/kafka/streams/errors/TaskMigratedException.java
index 5a284e48705aa..419543302c6a6 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/TaskMigratedException.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/TaskMigratedException.java
@@ -30,27 +30,35 @@ public class TaskMigratedException extends StreamsException {
 
     private final Task task;
 
-    public TaskMigratedException(final Task task) {
-        this(task, null);
+    // this is for unit test only
+    public TaskMigratedException() {
+        super("A task has been migrated unexpectedly", null);
+
+        this.task = null;
     }
 
     public TaskMigratedException(final Task task,
                                  final TopicPartition topicPartition,
                                  final long endOffset,
                                  final long pos) {
-        super(String.format("Log end offset of %s should not change while restoring: old end offset %d, current offset %d%n%s",
+        super(String.format("Log end offset of %s should not change while restoring: old end offset %d, current offset %d",
                             topicPartition,
                             endOffset,
-                            pos,
-                            task.toString("> ")),
+                            pos),
             null);
 
         this.task = task;
     }
 
+    public TaskMigratedException(final Task task) {
+        super(String.format("Task %s is unexpectedly closed during processing", task.id()), null);
+
+        this.task = task;
+    }
+
     public TaskMigratedException(final Task task,
                                  final Throwable throwable) {
-        super(task.toString(), throwable);
+        super(String.format("Client request for task %s has been fenced due to a rebalance", task.id()), throwable);
 
         this.task = task;
     }
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index ab96cce4c2fea..a7e3bcde4e84e 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -899,7 +899,7 @@ private void addRecordsToTasks(final ConsumerRecords<byte[], byte[]> records) {
             final StreamTask task = taskManager.activeTask(partition);
 
             if (task.isClosed()) {
-                log.warn("Stream task {} is already closed, probably because it got unexpectedly migrated to another thread already. " +
+                log.info("Stream task {} is already closed, probably because it got unexpectedly migrated to another thread already. " +
                         "Notifying the thread to trigger a new rebalance immediately.", task.id());
                 throw new TaskMigratedException(task);
             }
@@ -1032,7 +1032,7 @@ private void maybeUpdateStandbyTasks(final long now) {
                             final StandbyTask task = taskManager.standbyTask(partition);
 
                             if (task.isClosed()) {
-                                log.warn("Standby task {} is already closed, probably because it got unexpectly migrated to another thread already. " +
+                                log.info("Standby task {} is already closed, probably because it got unexpectedly migrated to another thread already. " +
                                         "Notifying the thread to trigger a new rebalance immediately.", task.id());
                                 throw new TaskMigratedException(task);
                             }
@@ -1065,7 +1065,7 @@ private void maybeUpdateStandbyTasks(final long now) {
                         }
 
                         if (task.isClosed()) {
-                            log.warn("Standby task {} is already closed, probably because it got unexpectedly migrated to another thread already. " +
+                            log.info("Standby task {} is already closed, probably because it got unexpectedly migrated to another thread already. " +
                                     "Notifying the thread to trigger a new rebalance immediately.", task.id());
                             throw new TaskMigratedException(task);
                         }
@@ -1084,7 +1084,7 @@ private void maybeUpdateStandbyTasks(final long now) {
                     final StandbyTask task = taskManager.standbyTask(partition);
 
                     if (task.isClosed()) {
-                        log.warn("Standby task {} is already closed, probably because it got unexpectly migrated to another thread already. " +
+                        log.info("Standby task {} is already closed, probably because it got unexpectedly migrated to another thread already. " +
                                 "Notifying the thread to trigger a new rebalance immediately.", task.id());
                         throw new TaskMigratedException(task);
                     }
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java
index fcd23220bac81..8a8d6255c46b5 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AssignedStreamsTasksTest.java
@@ -195,7 +195,7 @@ public void shouldCloseTaskOnSuspendWhenRuntimeException() {
     public void shouldCloseTaskOnSuspendIfTaskMigratedException() {
         mockTaskInitialization();
         t1.suspend();
-        EasyMock.expectLastCall().andThrow(new TaskMigratedException(t1));
+        EasyMock.expectLastCall().andThrow(new TaskMigratedException());
         t1.close(false, true);
         EasyMock.expectLastCall();
         EasyMock.replay(t1);
@@ -226,7 +226,7 @@ public void shouldCloseTaskOnResumeSuspendedIfTaskMigratedException() {
         mockRunningTaskSuspension();
         t1.resume();
         t1.initializeTopology();
-        EasyMock.expectLastCall().andThrow(new TaskMigratedException(t1));
+        EasyMock.expectLastCall().andThrow(new TaskMigratedException());
         t1.close(false, true);
         EasyMock.expectLastCall();
         EasyMock.replay(t1);
@@ -267,7 +267,7 @@ public void shouldCommitRunningTasks() {
     public void shouldCloseTaskOnCommitIfTaskMigratedException() {
         mockTaskInitialization();
         t1.commit();
-        EasyMock.expectLastCall().andThrow(new TaskMigratedException(t1));
+        EasyMock.expectLastCall().andThrow(new TaskMigratedException());
         t1.close(false, true);
         EasyMock.expectLastCall();
         EasyMock.replay(t1);
@@ -319,7 +319,7 @@ public void shouldCloseTaskOnMaybeCommitIfTaskMigratedException() {
         mockTaskInitialization();
         EasyMock.expect(t1.commitNeeded()).andReturn(true);
         t1.commit();
-        EasyMock.expectLastCall().andThrow(new TaskMigratedException(t1));
+        EasyMock.expectLastCall().andThrow(new TaskMigratedException());
         t1.close(false, true);
         EasyMock.expectLastCall();
         EasyMock.replay(t1);
@@ -338,7 +338,7 @@ public void shouldCloseTaskOnMaybeCommitIfTaskMigratedException() {
     public void shouldCloseTaskOnProcessesIfTaskMigratedException() {
         mockTaskInitialization();
         t1.process();
-        EasyMock.expectLastCall().andThrow(new TaskMigratedException(t1));
+        EasyMock.expectLastCall().andThrow(new TaskMigratedException());
         t1.close(false, true);
         EasyMock.expectLastCall();
         EasyMock.replay(t1);
@@ -370,7 +370,7 @@ public void shouldPunctuateRunningTasks() {
     public void shouldCloseTaskOnMaybePunctuateStreamTimeIfTaskMigratedException() {
         mockTaskInitialization();
         t1.maybePunctuateStreamTime();
-        EasyMock.expectLastCall().andThrow(new TaskMigratedException(t1));
+        EasyMock.expectLastCall().andThrow(new TaskMigratedException());
         t1.close(false, true);
         EasyMock.expectLastCall();
         EasyMock.replay(t1);
@@ -390,7 +390,7 @@ public void shouldCloseTaskOnMaybePunctuateSystemTimeIfTaskMigratedException() {
         mockTaskInitialization();
         EasyMock.expect(t1.maybePunctuateStreamTime()).andReturn(true);
         t1.maybePunctuateSystemTime();
-        EasyMock.expectLastCall().andThrow(new TaskMigratedException(t1));
+        EasyMock.expectLastCall().andThrow(new TaskMigratedException());
         t1.close(false, true);
         EasyMock.expectLastCall();
         EasyMock.replay(t1);