From d3fcefc7cb64bd329ec5740c579efe4f359f21aa Mon Sep 17 00:00:00 2001 From: Sriram Subramanian Date: Mon, 4 Mar 2013 16:03:28 -0800 Subject: [PATCH] tool changes --- .../PreferredReplicaLeaderElectionCommand.scala | 13 ++----------- .../kafka/admin/ReassignPartitionsCommand.scala | 8 -------- 2 files changed, 2 insertions(+), 19 deletions(-) diff --git a/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala b/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala index ebcf669..faf0e3e 100644 --- a/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala +++ b/core/src/main/scala/kafka/admin/PreferredReplicaLeaderElectionCommand.scala @@ -42,11 +42,11 @@ object PreferredReplicaLeaderElectionCommand extends Logging { val options = parser.parse(args : _*) - CommandLineUtils.checkRequiredArgs(parser, options, jsonFileOpt, zkConnectOpt) + CommandLineUtils.checkRequiredArgs(parser, options, zkConnectOpt) val jsonFile = options.valueOf(jsonFileOpt) val zkConnect = options.valueOf(zkConnectOpt) - val jsonString = Utils.readFileAsString(jsonFile) + val jsonString = if(jsonFile != "") Utils.readFileAsString(jsonFile) else "" var zkClient: ZkClient = null try { @@ -55,15 +55,6 @@ object PreferredReplicaLeaderElectionCommand extends Logging { if(jsonFile == "") ZkUtils.getAllPartitions(zkClient) else parsePreferredReplicaJsonData(jsonString) val preferredReplicaElectionCommand = new PreferredReplicaLeaderElectionCommand(zkClient, partitionsForPreferredReplicaElection) - // attach shutdown handler to catch control-c - Runtime.getRuntime().addShutdownHook(new Thread() { - override def run() = { - // delete the admin path so it can be retried - ZkUtils.deletePathRecursive(zkClient, ZkUtils.PreferredReplicaLeaderElectionPath) - zkClient.close() - } - }) - preferredReplicaElectionCommand.moveLeaderToPreferredReplica() println("Successfully started preferred replica election for partitions %s".format(partitionsForPreferredReplicaElection)) } catch { diff --git a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala index b2204b8..d3845cc 100644 --- a/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala +++ b/core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala @@ -71,14 +71,6 @@ object ReassignPartitionsCommand extends Logging { zkClient = new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer) val reassignPartitionsCommand = new ReassignPartitionsCommand(zkClient, partitionsToBeReassigned) - // attach shutdown handler to catch control-c - Runtime.getRuntime().addShutdownHook(new Thread() { - override def run() = { - // delete the admin path so it can be retried - ZkUtils.deletePathRecursive(zkClient, ZkUtils.ReassignPartitionsPath) - } - }) - if(reassignPartitionsCommand.reassignPartitions()) println("Successfully started reassignment of partitions %s".format(partitionsToBeReassigned)) else -- 1.7.1