diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
index 2843f3ef82c83..c306ca1c9f7ff 100755
--- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
@@ -20,14 +20,17 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.BufferedWriter;
 import java.io.Closeable;
 import java.io.DataOutput;
 import java.io.EOFException;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.OutputStreamWriter;
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.lang.reflect.Constructor;
@@ -570,6 +573,16 @@ public static String readFileAsString(String path) throws IOException {
         return Utils.readFileAsString(path, Charset.defaultCharset());
     }
 
+    public static void writeToFile(String path, String text) throws IOException {
+        BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(path), StandardCharsets.UTF_8));
+        try {
+            writer.write(text);
+        } finally {
+            writer.flush();
+            writer.close();
+        }
+    }
+
     /**
      * Check if the given ByteBuffer capacity
      * @param existingBuffer ByteBuffer capacity to check
diff --git a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
index e8aad7f167b4e..f0b02f9c9c5c2 100755
--- a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
+++ b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
@@ -16,6 +16,7 @@
  */
 package kafka.admin
 
+import java.io.File
 import java.util.Properties
 import java.util.concurrent.ExecutionException
 
@@ -193,10 +194,10 @@ object ReassignPartitionsCommand extends Logging {
     val interBrokerThrottle = opts.options.valueOf(opts.interBrokerThrottleOpt)
     val replicaAlterLogDirsThrottle = opts.options.valueOf(opts.replicaAlterLogDirsThrottleOpt)
     val timeoutMs = opts.options.valueOf(opts.timeoutOpt)
-    executeAssignment(zkUtils, adminClientOpt, reassignmentJsonString, Throttle(interBrokerThrottle, replicaAlterLogDirsThrottle), timeoutMs)
+    executeAssignment(zkUtils, adminClientOpt, reassignmentJsonString, reassignmentJsonFile, Throttle(interBrokerThrottle, replicaAlterLogDirsThrottle), timeoutMs)
   }
 
-  def executeAssignment(zkUtils: ZkUtils, adminClientOpt: Option[JAdminClient], reassignmentJsonString: String, throttle: Throttle, timeoutMs: Long = 10000L) {
+  def executeAssignment(zkUtils: ZkUtils, adminClientOpt: Option[JAdminClient], reassignmentJsonString: String, reassignmentJsonFile: String, throttle: Throttle, timeoutMs: Long = 10000L) {
     val (partitionAssignment, replicaAssignment) = parseAndValidate(zkUtils, reassignmentJsonString)
     val reassignPartitionsCommand = new ReassignPartitionsCommand(zkUtils, adminClientOpt, partitionAssignment.toMap, replicaAssignment)
 
@@ -205,7 +206,7 @@ object ReassignPartitionsCommand extends Logging {
       println("There is an existing assignment running.")
       reassignPartitionsCommand.maybeLimit(throttle)
     } else {
-      printCurrentAssignment(zkUtils, partitionAssignment.map(_._1.topic))
+      printCurrentAssignment(zkUtils, reassignmentJsonFile, partitionAssignment.map(_._1.topic))
       if (throttle.interBrokerLimit >= 0 || throttle.replicaAlterLogDirsLimit >= 0)
         println(String.format("Warning: You must run Verify periodically, until the reassignment completes, to ensure the throttle is removed. You can also alter the throttle by rerunning the Execute command passing a new value."))
       if (reassignPartitionsCommand.reassignPartitions(throttle, timeoutMs)) {
@@ -215,11 +216,23 @@ object ReassignPartitionsCommand extends Logging {
     }
   }
 
-  def printCurrentAssignment(zkUtils: ZkUtils, topics: Seq[String]): Unit = {
+  def printCurrentAssignment(zkUtils: ZkUtils, reassignmentJsonFile: String, topics: Seq[String]): Unit = {
     // before starting assignment, output the current replica assignment to facilitate rollback
     val currentPartitionReplicaAssignment = zkUtils.getReplicaAssignmentForTopics(topics)
-    println("Current partition replica assignment\n\n%s\n\nSave this to use as the --reassignment-json-file option during rollback"
-      .format(formatAsReassignmentJson(currentPartitionReplicaAssignment, Map.empty)))
+    println("Current partition replica assignment\n\n%s\n\n".format(formatAsReassignmentJson(currentPartitionReplicaAssignment, Map.empty)))
+    reassignmentJsonFile.isEmpty match {
+      case true =>
+        println("You can save this manually to use as the --reassignment-json-file option during rollback")
+      case false =>
+        val rollbackAssignmentJsonFile = new File(s"$reassignmentJsonFile.rollback")
+        rollbackAssignmentJsonFile.exists() match {
+          case true =>
+            println(s"File ${rollbackAssignmentJsonFile.getName} exists, you can save this manually to use as the --reassignment-json-file option during rollback")
+          case false =>
+            Utils.writeToFile(rollbackAssignmentJsonFile.getPath, ZkUtils.formatAsReassignmentJson(currentPartitionReplicaAssignment))
+            println(s"Saved this to ${rollbackAssignmentJsonFile.getName}, you can use it as the --reassignment-json-file option during rollback")
+        }
+    }
   }
 
   def formatAsReassignmentJson(partitionsToBeReassigned: Map[TopicAndPartition, Seq[Int]],
diff --git a/core/src/test/scala/other/kafka/ReplicationQuotasTestRig.scala b/core/src/test/scala/other/kafka/ReplicationQuotasTestRig.scala
index 389cb8f0ac396..bc9d241b160db 100644
--- a/core/src/test/scala/other/kafka/ReplicationQuotasTestRig.scala
+++ b/core/src/test/scala/other/kafka/ReplicationQuotasTestRig.scala
@@ -139,7 +139,7 @@ object ReplicationQuotasTestRig {
       val newAssignment = ReassignPartitionsCommand.generateAssignment(zkUtils, brokers, json(topicName), true)._1
 
       val start = System.currentTimeMillis()
-      ReassignPartitionsCommand.executeAssignment(zkUtils, None, ZkUtils.formatAsReassignmentJson(newAssignment), Throttle(config.throttle))
+      ReassignPartitionsCommand.executeAssignment(zkUtils, None, ZkUtils.formatAsReassignmentJson(newAssignment), "", Throttle(config.throttle))
 
       //Await completion
       waitForReassignmentToComplete()
diff --git a/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala b/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala
index 11f75e22c9321..525d890e2e4d3 100644
--- a/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala
@@ -90,7 +90,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
 
     //When we move the replica on 100 to broker 101
     val topicJson: String = s"""{"version":1,"partitions":[{"topic":"$topicName","partition":0,"replicas":[101],"log_dirs":["$expectedLogDir"]}]}"""
-    ReassignPartitionsCommand.executeAssignment(zkUtils, Some(adminClient), topicJson, NoThrottle)
+    ReassignPartitionsCommand.executeAssignment(zkUtils, Some(adminClient), topicJson, "", NoThrottle)
     waitForReassignmentToComplete()
 
     //Then the replica should be on 101
@@ -110,7 +110,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
 
     // When we execute an assignment that moves an existing replica to another log directory on the same broker
     val topicJson: String = s"""{"version":1,"partitions":[{"topic":"$topicName","partition":0,"replicas":[100],"log_dirs":["$expectedLogDir"]}]}"""
-    ReassignPartitionsCommand.executeAssignment(zkUtils, Some(adminClient), topicJson, NoThrottle)
+    ReassignPartitionsCommand.executeAssignment(zkUtils, Some(adminClient), topicJson, "", NoThrottle)
     val replica = new TopicPartitionReplica(topicName, 0, 100)
     TestUtils.waitUntilTrue(() => {
       expectedLogDir == adminClient.describeReplicaLogDirs(Collections.singleton(replica)).all().get.get(replica).getCurrentReplicaLogDir
@@ -144,7 +144,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
     // Before this reassignment, the replica already exists on broker 100 but does not exist on broker 102
     val newReplicaAssignment = Map(replica1 -> expectedLogDir1, replica2 -> expectedLogDir2)
     ReassignPartitionsCommand.executeAssignment(zkUtils, Some(adminClient),
-      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, newReplicaAssignment), NoThrottle)
+      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, newReplicaAssignment), "", NoThrottle)
     waitForReassignmentToComplete()
 
     // Then the replicas should span all three brokers
@@ -173,7 +173,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
     //When rebalancing
     val newAssignment = generateAssignment(zkUtils, Array(100, 101), json(topicName), true)._1
     ReassignPartitionsCommand.executeAssignment(zkUtils, None,
-      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, Map.empty), NoThrottle)
+      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, Map.empty), "", NoThrottle)
     waitForReassignmentToComplete()
 
     //Then replicas should only span the first two brokers
@@ -214,7 +214,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
 
     //When rebalancing
     ReassignPartitionsCommand.executeAssignment(zkUtils, Some(adminClient),
-      ReassignPartitionsCommand.formatAsReassignmentJson(proposed, proposedReplicaAssignment), NoThrottle)
+      ReassignPartitionsCommand.formatAsReassignmentJson(proposed, proposedReplicaAssignment), "", NoThrottle)
     waitForReassignmentToComplete()
 
     //Then the proposed changes should have been made
@@ -255,7 +255,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
 
     val start = System.currentTimeMillis()
     ReassignPartitionsCommand.executeAssignment(zkUtils, None,
-      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, Map.empty), initialThrottle)
+      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, Map.empty), "", initialThrottle)
 
     //Check throttle config. Should be throttling replica 0 on 100 and 102 only.
     checkThrottleConfigAddedToZK(initialThrottle.interBrokerLimit, servers, topicName, "0:100,0:101", "0:102")
@@ -308,7 +308,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
       TopicAndPartition("topic2", 2) -> Seq(103, 104)  //didn't move
     )
     ReassignPartitionsCommand.executeAssignment(zkUtils, None,
-      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, Map.empty), Throttle(throttle))
+      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, Map.empty), "", Throttle(throttle))
 
     //Check throttle config. Should be throttling specific replicas for each topic.
     checkThrottleConfigAddedToZK(throttle, servers, "topic1",
@@ -338,7 +338,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
     val newAssignment = generateAssignment(zkUtils, Array(101, 102), json(topicName), true)._1
 
     ReassignPartitionsCommand.executeAssignment(zkUtils, None,
-      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, Map.empty), Throttle(initialThrottle))
+      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, Map.empty), "", Throttle(initialThrottle))
 
     //Check throttle config
     checkThrottleConfigAddedToZK(initialThrottle, servers, topicName, "0:100,0:101", "0:102")
@@ -353,7 +353,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
     val newThrottle = initialThrottle * 1000
 
     ReassignPartitionsCommand.executeAssignment(zkUtils, None,
-      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, Map.empty), Throttle(newThrottle))
+      ReassignPartitionsCommand.formatAsReassignmentJson(newAssignment, Map.empty), "", Throttle(newThrottle))
 
     //Check throttle was changed
     checkThrottleConfigAddedToZK(newThrottle, servers, topicName, "0:100,0:101", "0:102")
@@ -380,7 +380,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
 
     //When we execute an assignment that includes an invalid partition (1:101 in this case)
     val topicJson = s"""{"version":1,"partitions":[{"topic":"$topicName","partition":1,"replicas":[101]}]}"""
-    ReassignPartitionsCommand.executeAssignment(zkUtils, None, topicJson, NoThrottle)
+    ReassignPartitionsCommand.executeAssignment(zkUtils, None, topicJson,"",  NoThrottle)
   }
 
   @Test(expected = classOf[AdminCommandFailedException])
@@ -391,7 +391,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
 
     //When we execute an assignment that specifies an empty replica list (0: empty list in this case)
     val topicJson = s"""{"version":1,"partitions":[{"topic":"$topicName","partition":0,"replicas":[]}]}"""
-    ReassignPartitionsCommand.executeAssignment(zkUtils, None, topicJson, NoThrottle)
+    ReassignPartitionsCommand.executeAssignment(zkUtils, None, topicJson, "", NoThrottle)
   }
 
   @Test(expected = classOf[AdminCommandFailedException])
@@ -402,7 +402,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
 
     //When we execute an assignment that specifies an invalid brokerID (102: invalid broker ID in this case)
     val topicJson = s"""{"version":1,"partitions":[{"topic":"$topicName","partition":0,"replicas":[101, 102]}]}"""
-    ReassignPartitionsCommand.executeAssignment(zkUtils, None, topicJson, NoThrottle)
+    ReassignPartitionsCommand.executeAssignment(zkUtils, None, topicJson, "", NoThrottle)
   }
 
   @Test(expected = classOf[AdminCommandFailedException])
@@ -414,7 +414,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
 
     // When we execute an assignment that specifies an invalid log directory
     val topicJson: String = s"""{"version":1,"partitions":[{"topic":"$topicName","partition":0,"replicas":[101],"log_dirs":["invalidDir"]}]}"""
-    ReassignPartitionsCommand.executeAssignment(zkUtils, Some(adminClient), topicJson, NoThrottle)
+    ReassignPartitionsCommand.executeAssignment(zkUtils, Some(adminClient), topicJson, "", NoThrottle)
   }
 
   @Test(expected = classOf[AdminCommandFailedException])
@@ -427,7 +427,7 @@ class ReassignPartitionsClusterTest extends ZooKeeperTestHarness with Logging {
 
     // When we execute an assignment whose length of replicas doesn't match that of replicas
     val topicJson: String = s"""{"version":1,"partitions":[{"topic":"$topicName","partition":0,"replicas":[101],"log_dirs":["$logDir", "$logDir"]}]}"""
-    ReassignPartitionsCommand.executeAssignment(zkUtils, Some(adminClient), topicJson, NoThrottle)
+    ReassignPartitionsCommand.executeAssignment(zkUtils, Some(adminClient), topicJson, "", NoThrottle)
   }
 
   @Test