Skip to content

KAFKA-5929: Save pre-assignment to file to avoid too long text to display when do topic partition reassign #3897

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
uncleGen wants to merge 3 commits into from
Closed

KAFKA-5929: Save pre-assignment to file to avoid too long text to display when do topic partition reassign #3897

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b5806c9
Save pre-assignment to file to avoid too long text to display when do…
uncleGen Sep 19, 2017
19d3f84
findbugs fix
uncleGen Sep 19, 2017
2465e68
rebase
uncleGen Nov 15, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/utils/Utils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down
25 changes: 19 additions & 6 deletions core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package kafka.admin

import java.io.File
import java.util.Properties
import java.util.concurrent.ExecutionException

Expand Down Expand Up @@ -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) {

@lindong28 lindong28 Nov 15, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

making reassignmentJsonFile an Option[String] is probably better than using empty to indicate that it is not used.

@uncleGen uncleGen Nov 16, 2017 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

get it, thanks for your review!

val (partitionAssignment, replicaAssignment) = parseAndValidate(zkUtils, reassignmentJsonString)
val reassignPartitionsCommand = new ReassignPartitionsCommand(zkUtils, adminClientOpt, partitionAssignment.toMap, replicaAssignment)

Expand All @@ -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)) {
Expand All @@ -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]],
Expand Down
2 changes: 1 addition & 1 deletion core/src/test/scala/other/kafka/ReplicationQuotasTestRig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
28 changes: 14 additions & 14 deletions core/src/test/scala/unit/kafka/admin/ReassignPartitionsClusterTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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")
Expand All @@ -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")
Expand All @@ -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])
Expand All @@ -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])
Expand All @@ -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])
Expand All @@ -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])
Expand All @@ -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
Expand Down