You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sr...@apache.org on 2014/02/25 09:27:11 UTC
[01/19] git commit: Move AddPartitions into TopicCommand
Repository: kafka
Updated Branches:
refs/heads/trunk 3955915a5 -> 00afb619c
Move AddPartitions into TopicCommand
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/c304a92a
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/c304a92a
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/c304a92a
Branch: refs/heads/trunk
Commit: c304a92a1ce39cc7b7e329fde0b97f93cb8caee3
Parents: 343a1e2
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Tue Oct 8 23:42:34 2013 -0700
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Tue Oct 8 23:42:34 2013 -0700
----------------------------------------------------------------------
.../kafka/admin/AddPartitionsCommand.scala | 127 -------------------
.../src/main/scala/kafka/admin/AdminUtils.scala | 57 +++++++++
.../main/scala/kafka/admin/TopicCommand.scala | 22 ++--
.../unit/kafka/admin/AddPartitionsTest.scala | 10 +-
4 files changed, 76 insertions(+), 140 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/c304a92a/core/src/main/scala/kafka/admin/AddPartitionsCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/AddPartitionsCommand.scala b/core/src/main/scala/kafka/admin/AddPartitionsCommand.scala
deleted file mode 100644
index c74d9c2..0000000
--- a/core/src/main/scala/kafka/admin/AddPartitionsCommand.scala
+++ /dev/null
@@ -1,127 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kafka.admin
-
-import joptsimple.OptionParser
-import kafka.utils._
-import org.I0Itec.zkclient.ZkClient
-import scala.collection.mutable
-import kafka.common.TopicAndPartition
-
-object AddPartitionsCommand extends Logging {
-
- def main(args: Array[String]): Unit = {
- val parser = new OptionParser
- val topicOpt = parser.accepts("topic", "REQUIRED: The topic for which partitions need to be added.")
- .withRequiredArg
- .describedAs("topic")
- .ofType(classOf[String])
- val zkConnectOpt = parser.accepts("zookeeper", "REQUIRED: The connection string for the zookeeper connection in the form host:port. " +
- "Multiple URLS can be given to allow fail-over.")
- .withRequiredArg
- .describedAs("urls")
- .ofType(classOf[String])
- val nPartitionsOpt = parser.accepts("partition", "REQUIRED: Number of partitions to add to the topic")
- .withRequiredArg
- .describedAs("# of partitions")
- .ofType(classOf[java.lang.Integer])
- val replicaAssignmentOpt = parser.accepts("replica-assignment-list", "For manually assigning replicas to brokers for the new partitions")
- .withRequiredArg
- .describedAs("broker_id_for_part1_replica1 : broker_id_for_part1_replica2, " +
- "broker_id_for_part2_replica1 : broker_id_for_part2_replica2, ...")
- .ofType(classOf[String])
- .defaultsTo("")
-
- val options = parser.parse(args : _*)
-
- for(arg <- List(topicOpt, zkConnectOpt, nPartitionsOpt)) {
- if(!options.has(arg)) {
- System.err.println("***Please note that this tool can only be used to add partitions when data for a topic does not use a key.***\n" +
- "Missing required argument. " + " \"" + arg + "\"")
- parser.printHelpOn(System.err)
- System.exit(1)
- }
- }
-
- val topic = options.valueOf(topicOpt)
- val zkConnect = options.valueOf(zkConnectOpt)
- val nPartitions = options.valueOf(nPartitionsOpt).intValue
- val replicaAssignmentStr = options.valueOf(replicaAssignmentOpt)
- var zkClient: ZkClient = null
- try {
- zkClient = new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer)
- addPartitions(zkClient, topic, nPartitions, replicaAssignmentStr)
- println("adding partitions succeeded!")
- } catch {
- case e: Throwable =>
- println("adding partitions failed because of " + e.getMessage)
- println(Utils.stackTrace(e))
- } finally {
- if (zkClient != null)
- zkClient.close()
- }
- }
-
- def addPartitions(zkClient: ZkClient, topic: String, numPartitions: Int = 1, replicaAssignmentStr: String = "") {
- val existingPartitionsReplicaList = ZkUtils.getReplicaAssignmentForTopics(zkClient, List(topic))
- if (existingPartitionsReplicaList.size == 0)
- throw new AdminOperationException("The topic %s does not exist".format(topic))
-
- val existingReplicaList = existingPartitionsReplicaList.get(TopicAndPartition(topic,0)).get
-
- // create the new partition replication list
- val brokerList = ZkUtils.getSortedBrokerList(zkClient)
- val newPartitionReplicaList = if (replicaAssignmentStr == "")
- AdminUtils.assignReplicasToBrokers(brokerList, numPartitions, existingReplicaList.size, existingReplicaList.head, existingPartitionsReplicaList.size)
- else
- getManualReplicaAssignment(replicaAssignmentStr, brokerList.toSet, existingPartitionsReplicaList.size)
-
- // check if manual assignment has the right replication factor
- val unmatchedRepFactorList = newPartitionReplicaList.values.filter(p => (p.size != existingReplicaList.size))
- if (unmatchedRepFactorList.size != 0)
- throw new AdminOperationException("The replication factor in manual replication assignment " +
- " is not equal to the existing replication factor for the topic " + existingReplicaList.size)
-
- info("Add partition list for %s is %s".format(topic, newPartitionReplicaList))
- val partitionReplicaList = existingPartitionsReplicaList.map(p => p._1.partition -> p._2)
- // add the new list
- partitionReplicaList ++= newPartitionReplicaList
- AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkClient, topic, partitionReplicaList, update = true)
- }
-
- def getManualReplicaAssignment(replicaAssignmentList: String, availableBrokerList: Set[Int], startPartitionId: Int): Map[Int, List[Int]] = {
- val partitionList = replicaAssignmentList.split(",")
- val ret = new mutable.HashMap[Int, List[Int]]()
- var partitionId = startPartitionId
- for (i <- 0 until partitionList.size) {
- val brokerList = partitionList(i).split(":").map(s => s.trim().toInt)
- if (brokerList.size <= 0)
- throw new AdminOperationException("replication factor must be larger than 0")
- if (brokerList.size != brokerList.toSet.size)
- throw new AdminOperationException("duplicate brokers in replica assignment: " + brokerList)
- if (!brokerList.toSet.subsetOf(availableBrokerList))
- throw new AdminOperationException("some specified brokers not available. specified brokers: " + brokerList.toString +
- "available broker:" + availableBrokerList.toString)
- ret.put(partitionId, brokerList.toList)
- if (ret(partitionId).size != ret(startPartitionId).size)
- throw new AdminOperationException("partition " + i + " has different replication factor: " + brokerList)
- partitionId = partitionId + 1
- }
- ret.toMap
- }
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/c304a92a/core/src/main/scala/kafka/admin/AdminUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/AdminUtils.scala b/core/src/main/scala/kafka/admin/AdminUtils.scala
index 6560fc6..1360955 100644
--- a/core/src/main/scala/kafka/admin/AdminUtils.scala
+++ b/core/src/main/scala/kafka/admin/AdminUtils.scala
@@ -29,6 +29,11 @@ import scala.collection._
import mutable.ListBuffer
import scala.collection.mutable
import kafka.common._
+import scala.Predef._
+import collection.Map
+import scala.Some
+import collection.Set
+import kafka.common.TopicAndPartition
object AdminUtils extends Logging {
val rand = new Random
@@ -82,6 +87,58 @@ object AdminUtils extends Logging {
}
ret.toMap
}
+
+ def addPartitions(zkClient: ZkClient, topic: String, numPartitions: Int = 1, replicaAssignmentStr: String = "") {
+ val existingPartitionsReplicaList = ZkUtils.getReplicaAssignmentForTopics(zkClient, List(topic))
+ if (existingPartitionsReplicaList.size == 0)
+ throw new AdminOperationException("The topic %s does not exist".format(topic))
+
+ val existingReplicaList = existingPartitionsReplicaList.head._2
+ val partitionsToAdd = numPartitions - existingPartitionsReplicaList.size
+ if (partitionsToAdd <= 0)
+ throw new AdminOperationException("The number of partitions for a topic can only be increased")
+
+ // create the new partition replication list
+ val brokerList = ZkUtils.getSortedBrokerList(zkClient)
+ val newPartitionReplicaList = if (replicaAssignmentStr == "")
+ AdminUtils.assignReplicasToBrokers(brokerList, partitionsToAdd, existingReplicaList.size, existingReplicaList.head, existingPartitionsReplicaList.size)
+ else
+ getManualReplicaAssignment(replicaAssignmentStr, brokerList.toSet, existingPartitionsReplicaList.size)
+
+ // check if manual assignment has the right replication factor
+ val unmatchedRepFactorList = newPartitionReplicaList.values.filter(p => (p.size != existingReplicaList.size))
+ if (unmatchedRepFactorList.size != 0)
+ throw new AdminOperationException("The replication factor in manual replication assignment " +
+ " is not equal to the existing replication factor for the topic " + existingReplicaList.size)
+
+ info("Add partition list for %s is %s".format(topic, newPartitionReplicaList))
+ val partitionReplicaList = existingPartitionsReplicaList.map(p => p._1.partition -> p._2)
+ // add the new list
+ partitionReplicaList ++= newPartitionReplicaList
+ AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkClient, topic, partitionReplicaList, update = true)
+ }
+
+ def getManualReplicaAssignment(replicaAssignmentList: String, availableBrokerList: Set[Int], startPartitionId: Int): Map[Int, List[Int]] = {
+ var partitionList = replicaAssignmentList.split(",")
+ val ret = new mutable.HashMap[Int, List[Int]]()
+ var partitionId = startPartitionId
+ partitionList = partitionList.takeRight(partitionList.size - partitionId)
+ for (i <- 0 until partitionList.size) {
+ val brokerList = partitionList(i).split(":").map(s => s.trim().toInt)
+ if (brokerList.size <= 0)
+ throw new AdminOperationException("replication factor must be larger than 0")
+ if (brokerList.size != brokerList.toSet.size)
+ throw new AdminOperationException("duplicate brokers in replica assignment: " + brokerList)
+ if (!brokerList.toSet.subsetOf(availableBrokerList))
+ throw new AdminOperationException("some specified brokers not available. specified brokers: " + brokerList.toString +
+ "available broker:" + availableBrokerList.toString)
+ ret.put(partitionId, brokerList.toList)
+ if (ret(partitionId).size != ret(startPartitionId).size)
+ throw new AdminOperationException("partition " + i + " has different replication factor: " + brokerList)
+ partitionId = partitionId + 1
+ }
+ ret.toMap
+ }
def deleteTopic(zkClient: ZkClient, topic: String) {
zkClient.deleteRecursive(ZkUtils.getTopicPath(topic))
http://git-wip-us.apache.org/repos/asf/kafka/blob/c304a92a/core/src/main/scala/kafka/admin/TopicCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala
index 06bbd37..ee940e9 100644
--- a/core/src/main/scala/kafka/admin/TopicCommand.scala
+++ b/core/src/main/scala/kafka/admin/TopicCommand.scala
@@ -78,16 +78,21 @@ object TopicCommand {
def alterTopic(zkClient: ZkClient, opts: TopicCommandOptions) {
CommandLineUtils.checkRequiredArgs(opts.parser, opts.options, opts.topicOpt)
- val topics = opts.options.valuesOf(opts.topicOpt)
- val configs = parseTopicConfigs(opts)
- if(opts.options.has(opts.partitionsOpt))
- Utils.croak("Changing the number of partitions is not supported.")
- if(opts.options.has(opts.replicationFactorOpt))
- Utils.croak("Changing the replication factor is not supported.")
- for(topic <- topics) {
+ val topic = opts.options.valueOf(opts.topicOpt)
+ if(opts.options.has(opts.configOpt)) {
+ val configs = parseTopicConfigs(opts)
AdminUtils.changeTopicConfig(zkClient, topic, configs)
println("Updated config for topic \"%s\".".format(topic))
}
+ if(opts.options.has(opts.partitionsOpt)) {
+ println("partitions can only be added when topic has no key")
+ val nPartitions = opts.options.valueOf(opts.partitionsOpt).intValue
+ val replicaAssignmentStr = opts.options.valueOf(opts.replicaAssignmentOpt)
+ AdminUtils.addPartitions(zkClient, topic, nPartitions, replicaAssignmentStr)
+ println("adding partitions succeeded!")
+ }
+ if(opts.options.has(opts.replicationFactorOpt))
+ Utils.croak("Changing the replication factor is not supported.")
}
def deleteTopic(zkClient: ZkClient, opts: TopicCommandOptions) {
@@ -182,7 +187,8 @@ object TopicCommand {
.withRequiredArg
.describedAs("name=value")
.ofType(classOf[String])
- val partitionsOpt = parser.accepts("partitions", "The number of partitions for the topic being created.")
+ val partitionsOpt = parser.accepts("partitions", "The number of partitions for the topic being created or " +
+ "altered (Partitions can only be added for a topic which has no key. Partitions cannot be decreased")
.withRequiredArg
.describedAs("# of partitions")
.ofType(classOf[java.lang.Integer])
http://git-wip-us.apache.org/repos/asf/kafka/blob/c304a92a/core/src/test/scala/unit/kafka/admin/AddPartitionsTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/admin/AddPartitionsTest.scala b/core/src/test/scala/unit/kafka/admin/AddPartitionsTest.scala
index 09254cc..115e203 100644
--- a/core/src/test/scala/unit/kafka/admin/AddPartitionsTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/AddPartitionsTest.scala
@@ -100,7 +100,7 @@ class AddPartitionsTest extends JUnit3Suite with ZooKeeperTestHarness {
def testTopicDoesNotExist {
try {
- AddPartitionsCommand.addPartitions(zkClient, "Blah", 1)
+ AdminUtils.addPartitions(zkClient, "Blah", 1)
fail("Topic should not exist")
} catch {
case e: AdminOperationException => //this is good
@@ -110,7 +110,7 @@ class AddPartitionsTest extends JUnit3Suite with ZooKeeperTestHarness {
def testWrongReplicaCount {
try {
- AddPartitionsCommand.addPartitions(zkClient, topic1, 2, "0:1:2")
+ AdminUtils.addPartitions(zkClient, topic1, 2, "0:1,0:1:2")
fail("Add partitions should fail")
} catch {
case e: AdminOperationException => //this is good
@@ -119,7 +119,7 @@ class AddPartitionsTest extends JUnit3Suite with ZooKeeperTestHarness {
}
def testIncrementPartitions {
- AddPartitionsCommand.addPartitions(zkClient, topic1, 2)
+ AdminUtils.addPartitions(zkClient, topic1, 3)
// wait until leader is elected
var leader1 = waitUntilLeaderIsElectedOrChanged(zkClient, topic1, 1, 500)
var leader2 = waitUntilLeaderIsElectedOrChanged(zkClient, topic1, 2, 500)
@@ -144,7 +144,7 @@ class AddPartitionsTest extends JUnit3Suite with ZooKeeperTestHarness {
}
def testManualAssignmentOfReplicas {
- AddPartitionsCommand.addPartitions(zkClient, topic2, 2, "0:1,2:3")
+ AdminUtils.addPartitions(zkClient, topic2, 3, "1:2,0:1,2:3")
// wait until leader is elected
var leader1 = waitUntilLeaderIsElectedOrChanged(zkClient, topic2, 1, 500)
var leader2 = waitUntilLeaderIsElectedOrChanged(zkClient, topic2, 2, 500)
@@ -170,7 +170,7 @@ class AddPartitionsTest extends JUnit3Suite with ZooKeeperTestHarness {
}
def testReplicaPlacement {
- AddPartitionsCommand.addPartitions(zkClient, topic3, 6)
+ AdminUtils.addPartitions(zkClient, topic3, 7)
// wait until leader is elected
var leader1 = waitUntilLeaderIsElectedOrChanged(zkClient, topic3, 1, 500)
var leader2 = waitUntilLeaderIsElectedOrChanged(zkClient, topic3, 2, 500)
[17/19] git commit: Merge branch 'trunk' of
http://git-wip-us.apache.org/repos/asf/kafka into trunk
Posted by sr...@apache.org.
Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into trunk
Conflicts:
core/src/main/scala/kafka/controller/KafkaController.scala
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/5de68ef4
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/5de68ef4
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/5de68ef4
Branch: refs/heads/trunk
Commit: 5de68ef4aef7812fd9f2d5e4fb6158bf753658e3
Parents: 2d848c5 993e1aa
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Mon Feb 24 01:50:02 2014 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Mon Feb 24 01:50:02 2014 -0800
----------------------------------------------------------------------
HEADER | 14 +
LICENSE | 31 -
README.md | 103 +-
bin/kafka-run-class.sh | 28 +-
bin/run-rat.sh | 35 -
build.gradle | 331 +++
clients/build.sbt | 11 -
.../producer/BufferExhaustedException.java | 17 -
.../java/kafka/clients/producer/Callback.java | 18 -
.../kafka/clients/producer/KafkaProducer.java | 286 ---
.../kafka/clients/producer/MockProducer.java | 199 --
.../java/kafka/clients/producer/Producer.java | 48 -
.../kafka/clients/producer/ProducerConfig.java | 130 --
.../kafka/clients/producer/ProducerRecord.java | 84 -
.../kafka/clients/producer/RecordMetadata.java | 39 -
.../clients/producer/internals/BufferPool.java | 223 --
.../internals/FutureRecordMetadata.java | 63 -
.../clients/producer/internals/Metadata.java | 120 -
.../clients/producer/internals/Partitioner.java | 55 -
.../internals/ProduceRequestResult.java | 81 -
.../producer/internals/RecordAccumulator.java | 234 --
.../clients/producer/internals/RecordBatch.java | 86 -
.../clients/producer/internals/Sender.java | 504 ----
.../clients/tools/ProducerPerformance.java | 65 -
clients/src/main/java/kafka/common/Cluster.java | 123 -
.../main/java/kafka/common/Configurable.java | 15 -
.../main/java/kafka/common/KafkaException.java | 26 -
clients/src/main/java/kafka/common/Metric.java | 23 -
clients/src/main/java/kafka/common/Node.java | 76 -
.../main/java/kafka/common/PartitionInfo.java | 58 -
.../main/java/kafka/common/TopicPartition.java | 61 -
.../kafka/common/config/AbstractConfig.java | 93 -
.../java/kafka/common/config/ConfigDef.java | 253 --
.../kafka/common/config/ConfigException.java | 24 -
.../java/kafka/common/errors/ApiException.java | 35 -
.../common/errors/CorruptRecordException.java | 23 -
.../errors/LeaderNotAvailableException.java | 19 -
.../kafka/common/errors/NetworkException.java | 23 -
.../errors/NotLeaderForPartitionException.java | 23 -
.../common/errors/OffsetMetadataTooLarge.java | 22 -
.../errors/OffsetOutOfRangeException.java | 22 -
.../common/errors/RecordTooLargeException.java | 23 -
.../kafka/common/errors/RetryableException.java | 31 -
.../kafka/common/errors/TimeoutException.java | 23 -
.../common/errors/UnknownServerException.java | 22 -
.../UnknownTopicOrPartitionException.java | 22 -
.../java/kafka/common/metrics/CompoundStat.java | 40 -
.../java/kafka/common/metrics/JmxReporter.java | 184 --
.../java/kafka/common/metrics/KafkaMetric.java | 55 -
.../java/kafka/common/metrics/Measurable.java | 16 -
.../kafka/common/metrics/MeasurableStat.java | 10 -
.../java/kafka/common/metrics/MetricConfig.java | 71 -
.../main/java/kafka/common/metrics/Metrics.java | 190 --
.../kafka/common/metrics/MetricsReporter.java | 27 -
.../main/java/kafka/common/metrics/Quota.java | 36 -
.../common/metrics/QuotaViolationException.java | 16 -
.../main/java/kafka/common/metrics/Sensor.java | 171 --
.../main/java/kafka/common/metrics/Stat.java | 16 -
.../java/kafka/common/metrics/stats/Avg.java | 33 -
.../java/kafka/common/metrics/stats/Count.java | 29 -
.../kafka/common/metrics/stats/Histogram.java | 137 --
.../java/kafka/common/metrics/stats/Max.java | 29 -
.../java/kafka/common/metrics/stats/Min.java | 29 -
.../kafka/common/metrics/stats/Percentile.java | 32 -
.../kafka/common/metrics/stats/Percentiles.java | 76 -
.../java/kafka/common/metrics/stats/Rate.java | 85 -
.../kafka/common/metrics/stats/SampledStat.java | 106 -
.../java/kafka/common/metrics/stats/Total.java | 31 -
.../kafka/common/network/ByteBufferReceive.java | 45 -
.../kafka/common/network/ByteBufferSend.java | 54 -
.../kafka/common/network/NetworkReceive.java | 74 -
.../java/kafka/common/network/NetworkSend.java | 26 -
.../main/java/kafka/common/network/Receive.java | 35 -
.../java/kafka/common/network/Selectable.java | 68 -
.../java/kafka/common/network/Selector.java | 349 ---
.../main/java/kafka/common/network/Send.java | 41 -
.../java/kafka/common/protocol/ApiKeys.java | 35 -
.../main/java/kafka/common/protocol/Errors.java | 97 -
.../java/kafka/common/protocol/ProtoUtils.java | 97 -
.../java/kafka/common/protocol/Protocol.java | 130 --
.../kafka/common/protocol/types/ArrayOf.java | 63 -
.../java/kafka/common/protocol/types/Field.java | 48 -
.../kafka/common/protocol/types/Schema.java | 134 --
.../common/protocol/types/SchemaException.java | 13 -
.../kafka/common/protocol/types/Struct.java | 227 --
.../java/kafka/common/protocol/types/Type.java | 216 --
.../kafka/common/record/CompressionType.java | 40 -
.../common/record/InvalidRecordException.java | 11 -
.../main/java/kafka/common/record/LogEntry.java | 28 -
.../java/kafka/common/record/MemoryRecords.java | 102 -
.../main/java/kafka/common/record/Record.java | 286 ---
.../main/java/kafka/common/record/Records.java | 29 -
.../kafka/common/requests/RequestHeader.java | 68 -
.../java/kafka/common/requests/RequestSend.java | 38 -
.../kafka/common/requests/ResponseHeader.java | 45 -
.../kafka/common/utils/AbstractIterator.java | 72 -
.../java/kafka/common/utils/CopyOnWriteMap.java | 130 --
.../src/main/java/kafka/common/utils/Crc32.java | 2169 ------------------
.../java/kafka/common/utils/KafkaThread.java | 18 -
.../java/kafka/common/utils/SystemTime.java | 26 -
.../src/main/java/kafka/common/utils/Time.java | 23 -
.../src/main/java/kafka/common/utils/Utils.java | 230 --
.../producer/BufferExhaustedException.java | 33 +
.../apache/kafka/clients/producer/Callback.java | 34 +
.../kafka/clients/producer/KafkaProducer.java | 303 +++
.../kafka/clients/producer/MockProducer.java | 216 ++
.../apache/kafka/clients/producer/Producer.java | 65 +
.../kafka/clients/producer/ProducerConfig.java | 170 ++
.../kafka/clients/producer/ProducerRecord.java | 100 +
.../kafka/clients/producer/RecordMetadata.java | 55 +
.../clients/producer/internals/BufferPool.java | 240 ++
.../internals/FutureRecordMetadata.java | 80 +
.../clients/producer/internals/Metadata.java | 139 ++
.../clients/producer/internals/Partitioner.java | 72 +
.../internals/ProduceRequestResult.java | 98 +
.../producer/internals/RecordAccumulator.java | 250 ++
.../clients/producer/internals/RecordBatch.java | 99 +
.../clients/producer/internals/Sender.java | 691 ++++++
.../clients/tools/ProducerPerformance.java | 85 +
.../java/org/apache/kafka/common/Cluster.java | 120 +
.../org/apache/kafka/common/Configurable.java | 31 +
.../org/apache/kafka/common/KafkaException.java | 42 +
.../java/org/apache/kafka/common/Metric.java | 39 +
.../main/java/org/apache/kafka/common/Node.java | 92 +
.../org/apache/kafka/common/PartitionInfo.java | 74 +
.../org/apache/kafka/common/TopicPartition.java | 77 +
.../kafka/common/config/AbstractConfig.java | 110 +
.../apache/kafka/common/config/ConfigDef.java | 269 +++
.../kafka/common/config/ConfigException.java | 40 +
.../kafka/common/errors/ApiException.java | 51 +
.../common/errors/CorruptRecordException.java | 39 +
.../common/errors/InvalidMetadataException.java | 39 +
.../errors/LeaderNotAvailableException.java | 27 +
.../kafka/common/errors/NetworkException.java | 39 +
.../errors/NotLeaderForPartitionException.java | 38 +
.../common/errors/OffsetMetadataTooLarge.java | 37 +
.../errors/OffsetOutOfRangeException.java | 38 +
.../common/errors/RecordTooLargeException.java | 38 +
.../kafka/common/errors/RetriableException.java | 37 +
.../kafka/common/errors/TimeoutException.java | 38 +
.../common/errors/UnknownServerException.java | 39 +
.../UnknownTopicOrPartitionException.java | 37 +
.../kafka/common/metrics/CompoundStat.java | 56 +
.../kafka/common/metrics/JmxReporter.java | 201 ++
.../kafka/common/metrics/KafkaMetric.java | 71 +
.../apache/kafka/common/metrics/Measurable.java | 32 +
.../kafka/common/metrics/MeasurableStat.java | 26 +
.../kafka/common/metrics/MetricConfig.java | 87 +
.../apache/kafka/common/metrics/Metrics.java | 207 ++
.../kafka/common/metrics/MetricsReporter.java | 43 +
.../org/apache/kafka/common/metrics/Quota.java | 52 +
.../common/metrics/QuotaViolationException.java | 32 +
.../org/apache/kafka/common/metrics/Sensor.java | 188 ++
.../org/apache/kafka/common/metrics/Stat.java | 32 +
.../apache/kafka/common/metrics/stats/Avg.java | 50 +
.../kafka/common/metrics/stats/Count.java | 46 +
.../kafka/common/metrics/stats/Histogram.java | 157 ++
.../apache/kafka/common/metrics/stats/Max.java | 46 +
.../apache/kafka/common/metrics/stats/Min.java | 46 +
.../kafka/common/metrics/stats/Percentile.java | 48 +
.../kafka/common/metrics/stats/Percentiles.java | 121 +
.../apache/kafka/common/metrics/stats/Rate.java | 106 +
.../kafka/common/metrics/stats/SampledStat.java | 127 +
.../kafka/common/metrics/stats/Total.java | 47 +
.../kafka/common/network/ByteBufferReceive.java | 61 +
.../kafka/common/network/ByteBufferSend.java | 70 +
.../kafka/common/network/NetworkReceive.java | 90 +
.../kafka/common/network/NetworkSend.java | 42 +
.../apache/kafka/common/network/Receive.java | 51 +
.../apache/kafka/common/network/Selectable.java | 84 +
.../apache/kafka/common/network/Selector.java | 364 +++
.../org/apache/kafka/common/network/Send.java | 57 +
.../apache/kafka/common/protocol/ApiKeys.java | 51 +
.../apache/kafka/common/protocol/Errors.java | 114 +
.../kafka/common/protocol/ProtoUtils.java | 65 +
.../apache/kafka/common/protocol/Protocol.java | 147 ++
.../kafka/common/protocol/types/ArrayOf.java | 79 +
.../kafka/common/protocol/types/Field.java | 64 +
.../kafka/common/protocol/types/Schema.java | 156 ++
.../common/protocol/types/SchemaException.java | 29 +
.../kafka/common/protocol/types/Struct.java | 243 ++
.../kafka/common/protocol/types/Type.java | 232 ++
.../kafka/common/record/CompressionType.java | 56 +
.../common/record/InvalidRecordException.java | 27 +
.../apache/kafka/common/record/LogEntry.java | 44 +
.../kafka/common/record/MemoryRecords.java | 119 +
.../org/apache/kafka/common/record/Record.java | 303 +++
.../org/apache/kafka/common/record/Records.java | 45 +
.../kafka/common/requests/MetadataRequest.java | 25 +
.../kafka/common/requests/MetadataResponse.java | 77 +
.../kafka/common/requests/RequestHeader.java | 85 +
.../kafka/common/requests/RequestSend.java | 55 +
.../kafka/common/requests/ResponseHeader.java | 62 +
.../kafka/common/utils/AbstractIterator.java | 88 +
.../kafka/common/utils/CopyOnWriteMap.java | 146 ++
.../org/apache/kafka/common/utils/Crc32.java | 2169 ++++++++++++++++++
.../apache/kafka/common/utils/KafkaThread.java | 34 +
.../apache/kafka/common/utils/SystemTime.java | 42 +
.../org/apache/kafka/common/utils/Time.java | 39 +
.../org/apache/kafka/common/utils/Utils.java | 247 ++
.../clients/common/network/SelectorTest.java | 292 ---
.../kafka/clients/producer/BufferPoolTest.java | 170 --
.../kafka/clients/producer/MetadataTest.java | 49 -
.../clients/producer/MockProducerTest.java | 63 -
.../kafka/clients/producer/PartitionerTest.java | 54 -
.../clients/producer/RecordAccumulatorTest.java | 135 --
.../kafka/clients/producer/RecordSendTest.java | 78 -
.../java/kafka/clients/producer/SenderTest.java | 87 -
.../java/kafka/common/config/ConfigDefTest.java | 88 -
.../kafka/common/metrics/JmxReporterTest.java | 21 -
.../java/kafka/common/metrics/MetricsTest.java | 176 --
.../common/metrics/stats/HistogramTest.java | 56 -
.../types/ProtocolSerializationTest.java | 96 -
.../kafka/common/record/MemoryRecordsTest.java | 44 -
.../java/kafka/common/record/RecordTest.java | 87 -
.../common/utils/AbstractIteratorTest.java | 54 -
.../test/java/kafka/common/utils/MockTime.java | 28 -
.../src/test/java/kafka/test/MetricsBench.java | 38 -
.../test/java/kafka/test/Microbenchmarks.java | 143 --
.../src/test/java/kafka/test/MockSelector.java | 87 -
clients/src/test/java/kafka/test/TestUtils.java | 95 -
.../kafka/clients/producer/BufferPoolTest.java | 187 ++
.../kafka/clients/producer/MetadataTest.java | 65 +
.../clients/producer/MockProducerTest.java | 82 +
.../kafka/clients/producer/PartitionerTest.java | 81 +
.../clients/producer/RecordAccumulatorTest.java | 146 ++
.../kafka/clients/producer/RecordSendTest.java | 94 +
.../kafka/clients/producer/SenderTest.java | 207 ++
.../kafka/common/config/ConfigDefTest.java | 106 +
.../kafka/common/metrics/JmxReporterTest.java | 40 +
.../kafka/common/metrics/MetricsTest.java | 234 ++
.../common/metrics/stats/HistogramTest.java | 103 +
.../kafka/common/network/SelectorTest.java | 308 +++
.../types/ProtocolSerializationTest.java | 118 +
.../kafka/common/record/MemoryRecordsTest.java | 63 +
.../apache/kafka/common/record/RecordTest.java | 106 +
.../common/utils/AbstractIteratorTest.java | 71 +
.../org/apache/kafka/common/utils/MockTime.java | 46 +
.../org/apache/kafka/test/MetricsBench.java | 55 +
.../org/apache/kafka/test/Microbenchmarks.java | 160 ++
.../org/apache/kafka/test/MockSelector.java | 104 +
.../java/org/apache/kafka/test/TestUtils.java | 112 +
contrib/hadoop-consumer/build.sbt | 1 -
contrib/hadoop-producer/build.sbt | 1 -
core/build.sbt | 32 -
.../src/main/scala/kafka/admin/AdminUtils.scala | 3 +-
.../main/scala/kafka/admin/TopicCommand.scala | 135 +-
.../kafka/api/ControlledShutdownResponse.scala | 4 +-
.../scala/kafka/api/LeaderAndIsrRequest.scala | 1 +
.../scala/kafka/api/StopReplicaRequest.scala | 22 +-
.../scala/kafka/api/StopReplicaResponse.scala | 18 +-
.../scala/kafka/api/UpdateMetadataRequest.scala | 1 +
.../main/scala/kafka/cluster/Partition.scala | 36 +-
.../scala/kafka/consumer/SimpleConsumer.scala | 8 +-
core/src/main/scala/kafka/consumer/package.html | 18 +
.../controller/ControllerChannelManager.scala | 160 +-
.../kafka/controller/KafkaController.scala | 328 ++-
.../controller/PartitionLeaderSelector.scala | 16 +-
.../controller/PartitionStateMachine.scala | 139 +-
.../kafka/controller/ReplicaStateMachine.scala | 194 +-
.../kafka/controller/TopicDeletionManager.scala | 376 +++
core/src/main/scala/kafka/log/package.html | 18 +
core/src/main/scala/kafka/message/package.html | 18 +
.../scala/kafka/network/BlockingChannel.scala | 16 +-
core/src/main/scala/kafka/network/package.html | 18 +
.../scala/kafka/producer/SyncProducer.scala | 6 +-
.../src/main/scala/kafka/server/KafkaApis.scala | 57 +-
.../main/scala/kafka/server/KafkaConfig.scala | 2 +-
.../scala/kafka/server/KafkaHealthcheck.scala | 9 +-
.../main/scala/kafka/server/KafkaServer.scala | 6 +-
.../scala/kafka/server/OffsetCheckpoint.scala | 1 -
.../scala/kafka/server/ReplicaManager.scala | 146 +-
.../scala/kafka/server/TopicConfigManager.scala | 9 +-
.../kafka/server/ZookeeperLeaderElector.scala | 10 +-
core/src/main/scala/kafka/server/package.html | 18 +
.../kafka/tools/newproducer/MirrorMaker.scala | 184 ++
.../scala/kafka/utils/CommandLineUtils.scala | 24 +-
core/src/main/scala/kafka/utils/ZkUtils.scala | 20 +-
core/src/main/scala/kafka/utils/package.html | 1 -
.../kafka/api/ProducerSendTest.scala | 275 +++
.../scala/other/kafka/TestCrcPerformance.scala | 16 +
.../scala/other/kafka/TestEndToEndLatency.scala | 22 +-
.../test/scala/unit/kafka/admin/AdminTest.scala | 52 +-
.../unit/kafka/admin/DeleteTopicTest.scala | 452 ++++
.../api/RequestResponseSerializationTest.scala | 6 +-
.../kafka/integration/PrimitiveApiTest.scala | 140 +-
.../scala/unit/kafka/server/LogOffsetTest.scala | 4 +-
.../unit/kafka/server/LogRecoveryTest.scala | 2 +-
.../unit/kafka/server/OffsetCommitTest.scala | 32 +-
.../unit/kafka/server/SimpleFetchTest.scala | 11 +-
.../unit/kafka/utils/IteratorTemplateTest.scala | 16 +
.../test/scala/unit/kafka/utils/JsonTest.scala | 16 +
.../test/scala/unit/kafka/utils/TestUtils.scala | 42 +-
.../unit/kafka/zk/ZooKeeperTestHarness.scala | 4 +-
examples/build.sbt | 3 -
gradle.properties | 23 +
gradle/buildscript.gradle | 12 +
gradle/license.gradle | 9 +
gradle/wrapper/gradle-wrapper.jar | Bin 0 -> 49875 bytes
gradle/wrapper/gradle-wrapper.properties | 6 +
gradlew | 164 ++
lib/sbt-launch.jar | Bin 1103618 -> 0 bytes
perf/build.sbt | 1 -
.../scala/kafka/perf/ConsumerPerformance.scala | 105 +-
.../scala/kafka/perf/ProducerPerformance.scala | 216 +-
project/Build.scala | 152 --
project/build.properties | 17 -
project/build/KafkaProject.scala | 251 --
project/plugins.sbt | 9 -
sbt | 16 -
sbt.bat | 17 -
scala.gradle | 5 +
settings.gradle | 17 +
313 files changed, 17403 insertions(+), 13130 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/5de68ef4/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/kafka/controller/KafkaController.scala
index f4f00b2,00a1f98..f12ffc2
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@@ -554,7 -612,8 +612,8 @@@ class KafkaController(val config : Kafk
} catch {
case e: Throwable => error("Error completing preferred replica leader election for partitions %s".format(partitions.mkString(",")), e)
} finally {
- removePartitionsFromPreferredReplicaElection(partitions)
+ removePartitionsFromPreferredReplicaElection(partitions, isTriggeredByAutoRebalance)
+ deleteTopicManager.resumeDeletionForTopics(partitions.map(_.topic))
}
}
@@@ -1008,6 -1090,6 +1092,7 @@@
topicsNotInPreferredReplica =
topicAndPartitionsForBroker.filter {
case(topicPartition, replicas) => {
++ controllerContext.partitionLeadershipInfo.contains(topicPartition) &&
controllerContext.partitionLeadershipInfo(topicPartition).leaderAndIsr.leader != leaderBroker
}
}
@@@ -1020,16 -1102,26 +1105,19 @@@
// check ratio and if greater than desired ratio, trigger a rebalance for the topic partitions
// that need to be on this broker
if (imbalanceRatio > (config.leaderImbalancePerBrokerPercentage.toDouble / 100)) {
- inLock(controllerContext.controllerLock) {
- // do this check only if the broker is live and there are no partitions being reassigned currently
- // and preferred replica election is not in progress
- if (controllerContext.liveBrokerIds.contains(leaderBroker) &&
- controllerContext.partitionsBeingReassigned.size == 0 &&
- controllerContext.partitionsUndergoingPreferredReplicaElection.size == 0) {
- val zkPath = ZkUtils.PreferredReplicaLeaderElectionPath
- val partitionsList = topicsNotInPreferredReplica.keys.map(e => Map("topic" -> e.topic, "partition" -> e.partition))
- val jsonData = Json.encode(Map("version" -> 1, "partitions" -> partitionsList))
- try {
- ZkUtils.createPersistentPath(zkClient, zkPath, jsonData)
- info("Created preferred replica election path with %s".format(jsonData))
- } catch {
- case e2: ZkNodeExistsException =>
- val partitionsUndergoingPreferredReplicaElection =
- PreferredReplicaLeaderElectionCommand.parsePreferredReplicaElectionData(ZkUtils.readData(zkClient, zkPath)._1)
- error("Preferred replica leader election currently in progress for " +
- "%s. Aborting operation".format(partitionsUndergoingPreferredReplicaElection));
- case e3: Throwable =>
- error("Error while trying to auto rebalance topics %s".format(topicsNotInPreferredReplica.keys))
+ topicsNotInPreferredReplica.foreach {
+ case(topicPartition, replicas) => {
- controllerContext.controllerLock synchronized {
++ inLock(controllerContext.controllerLock) {
+ // do this check only if the broker is live and there are no partitions being reassigned currently
+ // and preferred replica election is not in progress
+ if (controllerContext.liveBrokerIds.contains(leaderBroker) &&
- controllerContext.partitionsBeingReassigned.size == 0 &&
- controllerContext.partitionsUndergoingPreferredReplicaElection.size == 0) {
++ controllerContext.partitionsBeingReassigned.size == 0 &&
++ controllerContext.partitionsUndergoingPreferredReplicaElection.size == 0 &&
++ !deleteTopicManager.isTopicQueuedUpForDeletion(topicPartition.topic) &&
++ !deleteTopicManager.isTopicDeletionInProgress(topicPartition.topic) &&
++ controllerContext.allTopics.contains(topicPartition.topic)) {
+ onPreferredReplicaElection(Set(topicPartition), false)
+ }
}
}
}
[10/19] git commit: Merge branch 'trunk' of
http://git-wip-us.apache.org/repos/asf/kafka into trunk
Posted by sr...@apache.org.
Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into trunk
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/7f32a1c2
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/7f32a1c2
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/7f32a1c2
Branch: refs/heads/trunk
Commit: 7f32a1c276f5a027870d7cc340cce34274d9a896
Parents: 033872b cd81d37
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Thu Nov 21 09:25:35 2013 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Thu Nov 21 09:25:35 2013 -0800
----------------------------------------------------------------------
.../src/main/scala/kafka/admin/AdminUtils.scala | 2 +-
.../PreferredReplicaLeaderElectionCommand.scala | 9 +-
.../main/scala/kafka/api/FetchResponse.scala | 4 -
.../scala/kafka/api/LeaderAndIsrRequest.scala | 6 +-
.../main/scala/kafka/api/OffsetResponse.scala | 6 +-
.../main/scala/kafka/consumer/TopicCount.scala | 23 +-
.../consumer/ZookeeperConsumerConnector.scala | 8 +-
.../kafka/controller/KafkaController.scala | 2 +-
.../main/scala/kafka/log/FileMessageSet.scala | 2 -
core/src/main/scala/kafka/log/Log.scala | 28 +-
core/src/main/scala/kafka/log/LogManager.scala | 2 -
core/src/main/scala/kafka/log/OffsetIndex.scala | 40 +-
.../kafka/server/ZookeeperLeaderElector.scala | 6 +-
.../kafka/tools/ReplicaVerificationTool.scala | 387 +++++++++++++++++++
core/src/main/scala/kafka/utils/Utils.scala | 59 ---
core/src/main/scala/kafka/utils/ZkUtils.scala | 33 +-
.../api/RequestResponseSerializationTest.scala | 8 +-
.../scala/unit/kafka/log/LogSegmentTest.scala | 5 +-
.../src/test/scala/unit/kafka/log/LogTest.scala | 22 +-
.../test/scala/unit/kafka/utils/TestUtils.scala | 10 +-
project/Build.scala | 2 +
21 files changed, 470 insertions(+), 194 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/7f32a1c2/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
[04/19] git commit: Merge branch 'trunk' of
http://git-wip-us.apache.org/repos/asf/kafka into trunk
Posted by sr...@apache.org.
Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into trunk
Conflicts:
core/src/main/scala/kafka/admin/AdminUtils.scala
core/src/main/scala/kafka/admin/TopicCommand.scala
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/793b6076
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/793b6076
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/793b6076
Branch: refs/heads/trunk
Commit: 793b60763c0bb073dd5cacf3117b34877f557fb8
Parents: 7c920e9 eedbea6
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Mon Nov 18 03:15:27 2013 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Mon Nov 18 03:15:27 2013 -0800
----------------------------------------------------------------------
DISCLAIMER | 15 --
bin/kafka-add-partitions.sh | 18 --
bin/kafka-console-consumer.sh | 2 +-
bin/kafka-console-producer.sh | 2 +-
bin/kafka-consumer-perf-test.sh | 2 +-
bin/kafka-preferred-replica-election.sh | 2 +-
bin/kafka-producer-perf-test.sh | 2 +-
bin/kafka-reassign-partitions.sh | 2 +-
bin/kafka-replay-log-producer.sh | 2 +-
bin/kafka-run-class.sh | 18 +-
bin/kafka-server-start.sh | 2 +-
bin/kafka-simple-consumer-perf-test.sh | 2 +-
bin/kafka-simple-consumer-shell.sh | 2 +-
bin/kafka-topics.sh | 2 +-
bin/zookeeper-server-start.sh | 2 +-
bin/zookeeper-shell.sh | 2 +-
config/server.properties | 15 +-
.../src/main/scala/kafka/admin/AdminUtils.scala | 30 ++-
.../kafka/admin/ReassignPartitionsCommand.scala | 14 +-
.../main/scala/kafka/admin/TopicCommand.scala | 47 +++-
.../main/scala/kafka/cluster/Partition.scala | 97 +++----
.../main/scala/kafka/common/ErrorMapping.scala | 1 +
.../common/NotAssignedReplicaException.scala | 23 ++
.../scala/kafka/common/TopicAndPartition.scala | 6 +
.../kafka/consumer/ConsumerFetcherManager.scala | 47 ++--
.../kafka/consumer/ConsumerFetcherThread.scala | 2 +-
.../consumer/ZookeeperConsumerConnector.scala | 6 +-
.../kafka/controller/KafkaController.scala | 127 +++++++--
.../kafka/controller/ReplicaStateMachine.scala | 2 +
core/src/main/scala/kafka/log/Log.scala | 34 ++-
core/src/main/scala/kafka/log/LogConfig.scala | 2 +-
core/src/main/scala/kafka/log/LogManager.scala | 22 +-
core/src/main/scala/kafka/log/LogSegment.scala | 4 +-
core/src/main/scala/kafka/log/OffsetIndex.scala | 111 +++++---
.../scala/kafka/network/RequestChannel.scala | 29 ++-
.../kafka/server/AbstractFetcherManager.scala | 48 ++--
.../kafka/server/AbstractFetcherThread.scala | 38 +--
.../src/main/scala/kafka/server/KafkaApis.scala | 2 +-
.../main/scala/kafka/server/KafkaConfig.scala | 26 +-
.../scala/kafka/server/KafkaHealthcheck.scala | 14 +-
.../kafka/server/KafkaRequestHandler.scala | 2 +-
.../main/scala/kafka/server/KafkaServer.scala | 4 +-
.../scala/kafka/server/OffsetCheckpoint.scala | 16 +-
.../scala/kafka/server/ReplicaManager.scala | 259 +++++++++++++------
.../scala/kafka/server/TopicConfigManager.scala | 1 -
.../scala/kafka/tools/DumpLogSegments.scala | 12 +-
.../main/scala/kafka/tools/GetOffsetShell.scala | 71 +++--
core/src/main/scala/kafka/utils/Os.scala | 23 ++
core/src/main/scala/kafka/utils/Utils.scala | 13 +
.../kafka/utils/VerifiableProperties.scala | 2 +-
core/src/main/scala/kafka/utils/ZkUtils.scala | 12 +-
.../src/test/scala/unit/kafka/log/LogTest.scala | 42 ++-
.../unit/kafka/network/SocketServerTest.scala | 18 +-
.../unit/kafka/producer/ProducerTest.scala | 12 +-
.../unit/kafka/server/AdvertiseBrokerTest.scala | 52 ++++
.../unit/kafka/server/KafkaConfigTest.scala | 97 +++++++
.../test/scala/unit/kafka/utils/UtilsTest.scala | 14 +
kafka-patch-review.py | 13 +-
project/Build.scala | 3 +-
.../0.7/bin/kafka-run-class.sh | 8 +-
.../0.7/config/test-log4j.properties | 68 +++++
system_test/utils/kafka_system_test_utils.py | 1 +
62 files changed, 1106 insertions(+), 461 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/793b6076/core/src/main/scala/kafka/admin/AdminUtils.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/kafka/admin/AdminUtils.scala
index 9a8b0c9,8ff4bd5..05fb5b5
--- a/core/src/main/scala/kafka/admin/AdminUtils.scala
+++ b/core/src/main/scala/kafka/admin/AdminUtils.scala
@@@ -33,7 -33,6 +33,10 @@@ import scala.Predef.
import collection.Map
import scala.Some
import collection.Set
++<<<<<<< HEAD
+import kafka.common.TopicAndPartition
++=======
++>>>>>>> eedbea6526986783257ad0e025c451a8ee3d9095
object AdminUtils extends Logging {
val rand = new Random
http://git-wip-us.apache.org/repos/asf/kafka/blob/793b6076/core/src/main/scala/kafka/admin/TopicCommand.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/kafka/admin/TopicCommand.scala
index 56f3177,3c08dee..c3eecdd
--- a/core/src/main/scala/kafka/admin/TopicCommand.scala
+++ b/core/src/main/scala/kafka/admin/TopicCommand.scala
@@@ -79,8 -79,13 +79,18 @@@ object TopicCommand
def alterTopic(zkClient: ZkClient, opts: TopicCommandOptions) {
CommandLineUtils.checkRequiredArgs(opts.parser, opts.options, opts.topicOpt)
val topic = opts.options.valueOf(opts.topicOpt)
++<<<<<<< HEAD
+ if(opts.options.has(opts.configOpt)) {
+ val configs = parseTopicConfigs(opts)
++=======
+ if(opts.options.has(opts.configOpt) || opts.options.has(opts.deleteConfigOpt)) {
+ val configsToBeAdded = parseTopicConfigsToBeAdded(opts)
+ val configsToBeDeleted = parseTopicConfigsToBeDeleted(opts)
+ // compile the final set of configs
+ val configs = AdminUtils.fetchTopicConfig(zkClient, topic)
+ configs.putAll(configsToBeAdded)
+ configsToBeDeleted.foreach(config => configs.remove(config))
++>>>>>>> eedbea6526986783257ad0e025c451a8ee3d9095
AdminUtils.changeTopicConfig(zkClient, topic, configs)
println("Updated config for topic \"%s\".".format(topic))
}
@@@ -188,6 -207,10 +212,13 @@@
.withRequiredArg
.describedAs("name=value")
.ofType(classOf[String])
++<<<<<<< HEAD
++=======
+ val deleteConfigOpt = parser.accepts("deleteConfig", "A topic configuration override to be removed for an existing topic")
+ .withRequiredArg
+ .describedAs("name")
+ .ofType(classOf[String])
++>>>>>>> eedbea6526986783257ad0e025c451a8ee3d9095
val partitionsOpt = parser.accepts("partitions", "The number of partitions for the topic being created or " +
"altered (WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected")
.withRequiredArg
[06/19] git commit: Add auto leader rebalance support
Posted by sr...@apache.org.
Add auto leader rebalance support
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/5bcb4183
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/5bcb4183
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/5bcb4183
Branch: refs/heads/trunk
Commit: 5bcb41835f58be142bb6ac7c3155dfc163a516b4
Parents: 39c5b75
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Tue Nov 19 17:03:39 2013 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Tue Nov 19 17:03:39 2013 -0800
----------------------------------------------------------------------
.../kafka/controller/KafkaController.scala | 67 ++++++++++++++++++--
.../main/scala/kafka/server/KafkaConfig.scala | 12 ++++
2 files changed, 74 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/5bcb4183/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index 88792c2..d9d47c8 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -28,7 +28,7 @@ import kafka.common._
import kafka.metrics.{KafkaTimer, KafkaMetricsGroup}
import kafka.server.{ZookeeperLeaderElector, KafkaConfig}
import kafka.utils.ZkUtils._
-import kafka.utils.{Json, Utils, ZkUtils, Logging}
+import kafka.utils._
import org.apache.zookeeper.Watcher.Event.KeeperState
import org.I0Itec.zkclient.{IZkDataListener, IZkStateListener, ZkClient}
import org.I0Itec.zkclient.exception.{ZkNodeExistsException, ZkNoNodeException}
@@ -36,6 +36,11 @@ import java.util.concurrent.atomic.AtomicInteger
import scala.Some
import kafka.common.TopicAndPartition
import org.apache.log4j.Logger
+import scala.Some
+import kafka.common.TopicAndPartition
+import kafka.controller.ReassignedPartitionsContext
+import kafka.controller.PartitionAndReplica
+import kafka.controller.LeaderIsrAndControllerEpoch
class ControllerContext(val zkClient: ZkClient,
val zkSessionTimeout: Int,
@@ -112,6 +117,9 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
private val replicaStateMachine = new ReplicaStateMachine(this)
private val controllerElector = new ZookeeperLeaderElector(controllerContext, ZkUtils.ControllerPath, onControllerFailover,
config.brokerId)
+ // have a separate scheduler for the controller to be able to start and stop independently of the
+ // kafka server
+ private val controllerScheduler = new KafkaScheduler(1)
val offlinePartitionSelector = new OfflinePartitionLeaderSelector(controllerContext)
private val reassignedPartitionLeaderSelector = new ReassignedPartitionLeaderSelector(controllerContext)
private val preferredReplicaPartitionLeaderSelector = new PreferredReplicaPartitionLeaderSelector(controllerContext)
@@ -250,6 +258,12 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
initializeAndMaybeTriggerPreferredReplicaElection()
/* send partition leadership info to all live brokers */
sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq)
+ if (config.autoLeaderRebalanceEnable) {
+ info("starting the partition rebalance scheduler")
+ controllerScheduler.startup()
+ controllerScheduler.schedule("partition-rebalance-thread", checkAndTriggerPartitionRebalance,
+ 5, config.leaderImbalanceCheckIntervalSeconds, TimeUnit.SECONDS)
+ }
}
else
info("Controller has been shut down, aborting startup/failover")
@@ -456,7 +470,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
}
}
- def onPreferredReplicaElection(partitions: Set[TopicAndPartition]) {
+ def onPreferredReplicaElection(partitions: Set[TopicAndPartition], updateZk: Boolean = true) {
info("Starting preferred replica leader election for partitions %s".format(partitions.mkString(",")))
try {
controllerContext.partitionsUndergoingPreferredReplicaElection ++= partitions
@@ -464,7 +478,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
} catch {
case e: Throwable => error("Error completing preferred replica leader election for partitions %s".format(partitions.mkString(",")), e)
} finally {
- removePartitionsFromPreferredReplicaElection(partitions)
+ removePartitionsFromPreferredReplicaElection(partitions, updateZk)
}
}
@@ -493,6 +507,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
isRunning = false
partitionStateMachine.shutdown()
replicaStateMachine.shutdown()
+ controllerScheduler.shutdown()
if(controllerContext.controllerChannelManager != null) {
controllerContext.controllerChannelManager.shutdown()
controllerContext.controllerChannelManager = null
@@ -731,7 +746,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
}
}
- def removePartitionsFromPreferredReplicaElection(partitionsToBeRemoved: Set[TopicAndPartition]) {
+ def removePartitionsFromPreferredReplicaElection(partitionsToBeRemoved: Set[TopicAndPartition], updateZK : Boolean) {
for(partition <- partitionsToBeRemoved) {
// check the status
val currentLeader = controllerContext.partitionLeadershipInfo(partition).leaderAndIsr.leader
@@ -742,7 +757,8 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
warn("Partition %s failed to complete preferred replica leader election. Leader is %d".format(partition, currentLeader))
}
}
- ZkUtils.deletePath(zkClient, ZkUtils.PreferredReplicaLeaderElectionPath)
+ if (updateZK)
+ ZkUtils.deletePath(zkClient, ZkUtils.PreferredReplicaLeaderElectionPath)
controllerContext.partitionsUndergoingPreferredReplicaElection --= partitionsToBeRemoved
}
@@ -898,6 +914,47 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
}
}
}
+
+ private def checkAndTriggerPartitionRebalance(): Unit = {
+ if (isActive()) {
+ info("checking need to trigger partition rebalance")
+ // get all the active brokers
+ var preferredReplicasForTopicsByBrokers: Map[Int, Map[TopicAndPartition, Seq[Int]]] = null;
+ controllerContext.controllerLock synchronized {
+ preferredReplicasForTopicsByBrokers = controllerContext.partitionReplicaAssignment.groupBy(_._2.head)
+ }
+ debug("preferred replicas by broker " + preferredReplicasForTopicsByBrokers)
+ // for each broker, check if a preferred replica election needs to be triggered
+ preferredReplicasForTopicsByBrokers.foreach( brokerInfo => {
+ var imbalanceRatio: Double = 0
+ var topicsNotInPreferredReplica: Map[TopicAndPartition, Seq[Int]] = null
+ controllerContext.controllerLock synchronized {
+ val brokerIds = controllerContext.liveBrokerIds
+ if (brokerIds.contains(brokerInfo._1) &&
+ controllerContext.partitionsBeingReassigned.size == 0) {
+ // do this check only if the broker is live and there are no partitions being reassigned currently
+ topicsNotInPreferredReplica =
+ brokerInfo._2.filter(item => controllerContext.partitionLeadershipInfo(item._1).leaderAndIsr.leader != brokerInfo._1);
+ debug("topics not in preferred replica " + topicsNotInPreferredReplica)
+ val totalTopicPartitionsForBroker = brokerInfo._2.size
+ val totalTopicPartitionsNotLedByBroker = topicsNotInPreferredReplica.size
+ imbalanceRatio = totalTopicPartitionsNotLedByBroker.toDouble / totalTopicPartitionsForBroker
+ info("leader imbalance ratio for broker %d is %f".format(brokerInfo._1, imbalanceRatio))
+ }
+ }
+ // check ratio and if greater than desired ratio, trigger a rebalance for the topics
+ // that need to be on this broker
+ if (imbalanceRatio > (config.leaderImbalancePerBrokerPercentage.toDouble / 100)) {
+ topicsNotInPreferredReplica.foreach(topicPartition => {
+ controllerContext.controllerLock synchronized {
+ onPreferredReplicaElection(Set(topicPartition._1), false)
+ }
+ })
+ }
+ }
+ )
+ }
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/kafka/blob/5bcb4183/core/src/main/scala/kafka/server/KafkaConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index b324344..921f456 100644
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -229,6 +229,18 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro
/* the purge interval (in number of requests) of the producer request purgatory */
val producerPurgatoryPurgeIntervalRequests = props.getInt("producer.purgatory.purge.interval.requests", 10000)
+ /* Enables auto leader balancing. A background thread checks and triggers leader
+ * balance if required at regular intervals */
+ val autoLeaderRebalanceEnable = props.getBoolean("auto.leader.rebalance.enable", false)
+
+ /* the ratio of leader imbalance allowed per broker. The controller would trigger a leader balance if it goes above
+ * this value per broker. The value is specified in percentage. */
+ val leaderImbalancePerBrokerPercentage = props.getInt("leader.imbalance.per.broker.percentage", 10)
+
+ /* the frequency with which the partition rebalance check is triggered by the controller */
+ val leaderImbalanceCheckIntervalSeconds = props.getInt("leader.imbalance.check.interval.seconds", 300)
+
+
/*********** Controlled shutdown configuration ***********/
/** Controlled shutdown can fail for multiple reasons. This determines the number of retries when such failure happens */
[07/19] git commit: fix merge conflicts
Posted by sr...@apache.org.
fix merge conflicts
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/09e807bd
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/09e807bd
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/09e807bd
Branch: refs/heads/trunk
Commit: 09e807bda103fe632e62312f036b850a59576f36
Parents: 5bcb418
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Tue Nov 19 17:07:43 2013 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Tue Nov 19 17:07:43 2013 -0800
----------------------------------------------------------------------
core/src/main/scala/kafka/admin/TopicCommand.scala | 8 --------
1 file changed, 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/09e807bd/core/src/main/scala/kafka/admin/TopicCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala
index c3eecdd..3c08dee 100644
--- a/core/src/main/scala/kafka/admin/TopicCommand.scala
+++ b/core/src/main/scala/kafka/admin/TopicCommand.scala
@@ -79,10 +79,6 @@ object TopicCommand {
def alterTopic(zkClient: ZkClient, opts: TopicCommandOptions) {
CommandLineUtils.checkRequiredArgs(opts.parser, opts.options, opts.topicOpt)
val topic = opts.options.valueOf(opts.topicOpt)
-<<<<<<< HEAD
- if(opts.options.has(opts.configOpt)) {
- val configs = parseTopicConfigs(opts)
-=======
if(opts.options.has(opts.configOpt) || opts.options.has(opts.deleteConfigOpt)) {
val configsToBeAdded = parseTopicConfigsToBeAdded(opts)
val configsToBeDeleted = parseTopicConfigsToBeDeleted(opts)
@@ -90,7 +86,6 @@ object TopicCommand {
val configs = AdminUtils.fetchTopicConfig(zkClient, topic)
configs.putAll(configsToBeAdded)
configsToBeDeleted.foreach(config => configs.remove(config))
->>>>>>> eedbea6526986783257ad0e025c451a8ee3d9095
AdminUtils.changeTopicConfig(zkClient, topic, configs)
println("Updated config for topic \"%s\".".format(topic))
}
@@ -212,13 +207,10 @@ object TopicCommand {
.withRequiredArg
.describedAs("name=value")
.ofType(classOf[String])
-<<<<<<< HEAD
-=======
val deleteConfigOpt = parser.accepts("deleteConfig", "A topic configuration override to be removed for an existing topic")
.withRequiredArg
.describedAs("name")
.ofType(classOf[String])
->>>>>>> eedbea6526986783257ad0e025c451a8ee3d9095
val partitionsOpt = parser.accepts("partitions", "The number of partitions for the topic being created or " +
"altered (WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected")
.withRequiredArg
[09/19] git commit: commit missing code
Posted by sr...@apache.org.
commit missing code
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/033872b3
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/033872b3
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/033872b3
Branch: refs/heads/trunk
Commit: 033872b316fd5a68d7463138d8199fb5d821f41b
Parents: a4cd17a
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Tue Nov 19 17:38:13 2013 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Tue Nov 19 17:38:13 2013 -0800
----------------------------------------------------------------------
core/src/main/scala/kafka/controller/KafkaController.scala | 9 +++++----
1 file changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/033872b3/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index 476ed86..e2ad682 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -114,7 +114,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
config.brokerId)
// have a separate scheduler for the controller to be able to start and stop independently of the
// kafka server
- private val controllerScheduler = new KafkaScheduler(1)
+ private val autoRebalanceScheduler = new KafkaScheduler(1)
val offlinePartitionSelector = new OfflinePartitionLeaderSelector(controllerContext)
private val reassignedPartitionLeaderSelector = new ReassignedPartitionLeaderSelector(controllerContext)
private val preferredReplicaPartitionLeaderSelector = new PreferredReplicaPartitionLeaderSelector(controllerContext)
@@ -255,8 +255,8 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq)
if (config.autoLeaderRebalanceEnable) {
info("starting the partition rebalance scheduler")
- controllerScheduler.startup()
- controllerScheduler.schedule("partition-rebalance-thread", checkAndTriggerPartitionRebalance,
+ autoRebalanceScheduler.startup()
+ autoRebalanceScheduler.schedule("partition-rebalance-thread", checkAndTriggerPartitionRebalance,
5, config.leaderImbalanceCheckIntervalSeconds, TimeUnit.SECONDS)
}
}
@@ -502,7 +502,8 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
isRunning = false
partitionStateMachine.shutdown()
replicaStateMachine.shutdown()
- controllerScheduler.shutdown()
+ if (config.autoLeaderRebalanceEnable)
+ autoRebalanceScheduler.shutdown()
if(controllerContext.controllerChannelManager != null) {
controllerContext.controllerChannelManager.shutdown()
controllerContext.controllerChannelManager = null
[08/19] git commit: some more changes
Posted by sr...@apache.org.
some more changes
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/a4cd17a3
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/a4cd17a3
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/a4cd17a3
Branch: refs/heads/trunk
Commit: a4cd17a37bf0f5ccae7b8f6a404c8b8c0de4dbad
Parents: 09e807b
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Tue Nov 19 17:15:01 2013 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Tue Nov 19 17:15:01 2013 -0800
----------------------------------------------------------------------
core/src/main/scala/kafka/admin/AdminUtils.scala | 4 ----
core/src/main/scala/kafka/controller/KafkaController.scala | 7 +------
2 files changed, 1 insertion(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/a4cd17a3/core/src/main/scala/kafka/admin/AdminUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/AdminUtils.scala b/core/src/main/scala/kafka/admin/AdminUtils.scala
index 05fb5b5..8ff4bd5 100644
--- a/core/src/main/scala/kafka/admin/AdminUtils.scala
+++ b/core/src/main/scala/kafka/admin/AdminUtils.scala
@@ -33,10 +33,6 @@ import scala.Predef._
import collection.Map
import scala.Some
import collection.Set
-<<<<<<< HEAD
-import kafka.common.TopicAndPartition
-=======
->>>>>>> eedbea6526986783257ad0e025c451a8ee3d9095
object AdminUtils extends Logging {
val rand = new Random
http://git-wip-us.apache.org/repos/asf/kafka/blob/a4cd17a3/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index d9d47c8..476ed86 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -28,7 +28,7 @@ import kafka.common._
import kafka.metrics.{KafkaTimer, KafkaMetricsGroup}
import kafka.server.{ZookeeperLeaderElector, KafkaConfig}
import kafka.utils.ZkUtils._
-import kafka.utils._
+import kafka.utils.{Json, Utils, ZkUtils, Logging, KafkaScheduler}
import org.apache.zookeeper.Watcher.Event.KeeperState
import org.I0Itec.zkclient.{IZkDataListener, IZkStateListener, ZkClient}
import org.I0Itec.zkclient.exception.{ZkNodeExistsException, ZkNoNodeException}
@@ -36,11 +36,6 @@ import java.util.concurrent.atomic.AtomicInteger
import scala.Some
import kafka.common.TopicAndPartition
import org.apache.log4j.Logger
-import scala.Some
-import kafka.common.TopicAndPartition
-import kafka.controller.ReassignedPartitionsContext
-import kafka.controller.PartitionAndReplica
-import kafka.controller.LeaderIsrAndControllerEpoch
class ControllerContext(val zkClient: ZkClient,
val zkSessionTimeout: Int,
[16/19] git commit: Merge branch 'trunk' of
http://git-wip-us.apache.org/repos/asf/kafka into trunk
Posted by sr...@apache.org.
Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into trunk
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/2d848c5a
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/2d848c5a
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/2d848c5a
Branch: refs/heads/trunk
Commit: 2d848c5ae0d10fc4d49ad9a3477b8c46f1c3dab7
Parents: 3f47356 253f86e
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Wed Feb 5 22:06:42 2014 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Wed Feb 5 22:06:42 2014 -0800
----------------------------------------------------------------------
bin/windows/kafka-run-class.bat | 2 +-
clients/build.sbt | 11 +
.../producer/BufferExhaustedException.java | 17 +
.../java/kafka/clients/producer/Callback.java | 18 +
.../kafka/clients/producer/KafkaProducer.java | 286 +++
.../kafka/clients/producer/MockProducer.java | 199 ++
.../java/kafka/clients/producer/Producer.java | 48 +
.../kafka/clients/producer/ProducerConfig.java | 130 ++
.../kafka/clients/producer/ProducerRecord.java | 84 +
.../kafka/clients/producer/RecordMetadata.java | 39 +
.../clients/producer/internals/BufferPool.java | 223 ++
.../internals/FutureRecordMetadata.java | 63 +
.../clients/producer/internals/Metadata.java | 120 +
.../clients/producer/internals/Partitioner.java | 55 +
.../internals/ProduceRequestResult.java | 81 +
.../producer/internals/RecordAccumulator.java | 234 ++
.../clients/producer/internals/RecordBatch.java | 86 +
.../clients/producer/internals/Sender.java | 504 ++++
.../clients/tools/ProducerPerformance.java | 65 +
clients/src/main/java/kafka/common/Cluster.java | 123 +
.../main/java/kafka/common/Configurable.java | 15 +
.../main/java/kafka/common/KafkaException.java | 26 +
clients/src/main/java/kafka/common/Metric.java | 23 +
clients/src/main/java/kafka/common/Node.java | 76 +
.../main/java/kafka/common/PartitionInfo.java | 58 +
.../main/java/kafka/common/TopicPartition.java | 61 +
.../kafka/common/config/AbstractConfig.java | 93 +
.../java/kafka/common/config/ConfigDef.java | 253 ++
.../kafka/common/config/ConfigException.java | 24 +
.../java/kafka/common/errors/ApiException.java | 35 +
.../common/errors/CorruptRecordException.java | 23 +
.../errors/LeaderNotAvailableException.java | 19 +
.../kafka/common/errors/NetworkException.java | 23 +
.../errors/NotLeaderForPartitionException.java | 23 +
.../common/errors/OffsetMetadataTooLarge.java | 22 +
.../errors/OffsetOutOfRangeException.java | 22 +
.../common/errors/RecordTooLargeException.java | 23 +
.../kafka/common/errors/RetryableException.java | 31 +
.../kafka/common/errors/TimeoutException.java | 23 +
.../common/errors/UnknownServerException.java | 22 +
.../UnknownTopicOrPartitionException.java | 22 +
.../java/kafka/common/metrics/CompoundStat.java | 40 +
.../java/kafka/common/metrics/JmxReporter.java | 184 ++
.../java/kafka/common/metrics/KafkaMetric.java | 55 +
.../java/kafka/common/metrics/Measurable.java | 16 +
.../kafka/common/metrics/MeasurableStat.java | 10 +
.../java/kafka/common/metrics/MetricConfig.java | 71 +
.../main/java/kafka/common/metrics/Metrics.java | 190 ++
.../kafka/common/metrics/MetricsReporter.java | 27 +
.../main/java/kafka/common/metrics/Quota.java | 36 +
.../common/metrics/QuotaViolationException.java | 16 +
.../main/java/kafka/common/metrics/Sensor.java | 171 ++
.../main/java/kafka/common/metrics/Stat.java | 16 +
.../java/kafka/common/metrics/stats/Avg.java | 33 +
.../java/kafka/common/metrics/stats/Count.java | 29 +
.../kafka/common/metrics/stats/Histogram.java | 137 ++
.../java/kafka/common/metrics/stats/Max.java | 29 +
.../java/kafka/common/metrics/stats/Min.java | 29 +
.../kafka/common/metrics/stats/Percentile.java | 32 +
.../kafka/common/metrics/stats/Percentiles.java | 76 +
.../java/kafka/common/metrics/stats/Rate.java | 85 +
.../kafka/common/metrics/stats/SampledStat.java | 106 +
.../java/kafka/common/metrics/stats/Total.java | 31 +
.../kafka/common/network/ByteBufferReceive.java | 45 +
.../kafka/common/network/ByteBufferSend.java | 54 +
.../kafka/common/network/NetworkReceive.java | 74 +
.../java/kafka/common/network/NetworkSend.java | 26 +
.../main/java/kafka/common/network/Receive.java | 35 +
.../java/kafka/common/network/Selectable.java | 68 +
.../java/kafka/common/network/Selector.java | 349 +++
.../main/java/kafka/common/network/Send.java | 41 +
.../java/kafka/common/protocol/ApiKeys.java | 35 +
.../main/java/kafka/common/protocol/Errors.java | 97 +
.../java/kafka/common/protocol/ProtoUtils.java | 97 +
.../java/kafka/common/protocol/Protocol.java | 130 ++
.../kafka/common/protocol/types/ArrayOf.java | 63 +
.../java/kafka/common/protocol/types/Field.java | 48 +
.../kafka/common/protocol/types/Schema.java | 134 ++
.../common/protocol/types/SchemaException.java | 13 +
.../kafka/common/protocol/types/Struct.java | 227 ++
.../java/kafka/common/protocol/types/Type.java | 216 ++
.../kafka/common/record/CompressionType.java | 40 +
.../common/record/InvalidRecordException.java | 11 +
.../main/java/kafka/common/record/LogEntry.java | 28 +
.../java/kafka/common/record/MemoryRecords.java | 102 +
.../main/java/kafka/common/record/Record.java | 286 +++
.../main/java/kafka/common/record/Records.java | 29 +
.../kafka/common/requests/RequestHeader.java | 68 +
.../java/kafka/common/requests/RequestSend.java | 38 +
.../kafka/common/requests/ResponseHeader.java | 45 +
.../kafka/common/utils/AbstractIterator.java | 72 +
.../java/kafka/common/utils/CopyOnWriteMap.java | 130 ++
.../src/main/java/kafka/common/utils/Crc32.java | 2169 ++++++++++++++++++
.../java/kafka/common/utils/KafkaThread.java | 18 +
.../java/kafka/common/utils/SystemTime.java | 26 +
.../src/main/java/kafka/common/utils/Time.java | 23 +
.../src/main/java/kafka/common/utils/Utils.java | 230 ++
.../clients/common/network/SelectorTest.java | 292 +++
.../kafka/clients/producer/BufferPoolTest.java | 170 ++
.../kafka/clients/producer/MetadataTest.java | 49 +
.../clients/producer/MockProducerTest.java | 63 +
.../kafka/clients/producer/PartitionerTest.java | 54 +
.../clients/producer/RecordAccumulatorTest.java | 135 ++
.../kafka/clients/producer/RecordSendTest.java | 78 +
.../java/kafka/clients/producer/SenderTest.java | 87 +
.../java/kafka/common/config/ConfigDefTest.java | 88 +
.../kafka/common/metrics/JmxReporterTest.java | 21 +
.../java/kafka/common/metrics/MetricsTest.java | 176 ++
.../common/metrics/stats/HistogramTest.java | 56 +
.../types/ProtocolSerializationTest.java | 96 +
.../kafka/common/record/MemoryRecordsTest.java | 44 +
.../java/kafka/common/record/RecordTest.java | 87 +
.../common/utils/AbstractIteratorTest.java | 54 +
.../test/java/kafka/common/utils/MockTime.java | 28 +
.../src/test/java/kafka/test/MetricsBench.java | 38 +
.../test/java/kafka/test/Microbenchmarks.java | 143 ++
.../src/test/java/kafka/test/MockSelector.java | 87 +
clients/src/test/java/kafka/test/TestUtils.java | 95 +
contrib/hadoop-consumer/README | 2 +-
core/build.sbt | 2 +-
.../scala/kafka/consumer/ConsumerConfig.scala | 4 +
.../kafka/consumer/ConsumerFetcherManager.scala | 2 +-
.../scala/kafka/consumer/SimpleConsumer.scala | 9 +-
.../test/scala/unit/kafka/utils/TestUtils.scala | 1 +
.../scala/kafka/perf/ConsumerPerformance.scala | 6 +
project/Build.scala | 1 +
project/build/KafkaProject.scala | 2 +-
.../0.7/bin/kafka-run-class.sh | 2 +-
128 files changed, 11752 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
[15/19] git commit: Merge branch 'trunk' of
http://git-wip-us.apache.org/repos/asf/kafka into trunk
Posted by sr...@apache.org.
Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into trunk
Conflicts:
core/src/main/scala/kafka/controller/KafkaController.scala
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/3f473566
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/3f473566
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/3f473566
Branch: refs/heads/trunk
Commit: 3f473566ab2ef51f17034dc02f632df5d38fe307
Parents: 72610d1 26a02c3
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Mon Jan 27 13:25:06 2014 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Mon Jan 27 13:25:06 2014 -0800
----------------------------------------------------------------------
bin/zookeeper-shell.sh | 6 +-
.../main/scala/kafka/admin/TopicCommand.scala | 10 +-
.../kafka/api/ControlledShutdownRequest.scala | 16 +-
.../kafka/api/ControlledShutdownResponse.scala | 3 +
.../src/main/scala/kafka/api/FetchRequest.scala | 26 ++-
.../scala/kafka/api/LeaderAndIsrRequest.scala | 23 +-
.../scala/kafka/api/LeaderAndIsrResponse.scala | 3 +
.../scala/kafka/api/OffsetCommitRequest.scala | 16 ++
.../scala/kafka/api/OffsetCommitResponse.scala | 3 +
.../scala/kafka/api/OffsetFetchRequest.scala | 18 +-
.../scala/kafka/api/OffsetFetchResponse.scala | 4 +-
.../main/scala/kafka/api/OffsetRequest.scala | 23 +-
.../main/scala/kafka/api/OffsetResponse.scala | 1 +
.../main/scala/kafka/api/ProducerRequest.scala | 24 ++-
.../main/scala/kafka/api/ProducerResponse.scala | 2 +
.../scala/kafka/api/RequestOrResponse.scala | 8 +-
.../scala/kafka/api/StopReplicaRequest.scala | 25 ++-
.../scala/kafka/api/StopReplicaResponse.scala | 2 +
.../scala/kafka/api/TopicMetadataRequest.scala | 21 +-
.../scala/kafka/api/TopicMetadataResponse.scala | 2 +
.../scala/kafka/api/UpdateMetadataRequest.scala | 19 +-
.../kafka/api/UpdateMetadataResponse.scala | 2 +
.../main/scala/kafka/cluster/Partition.scala | 19 +-
.../common/LogCleaningAbortedException.scala | 24 +++
.../common/OptimisticLockFailureException.scala | 23 --
.../kafka/common/ThreadShutdownException.scala | 24 +++
.../controller/ControllerChannelManager.scala | 53 +++--
.../kafka/controller/KafkaController.scala | 198 +++++++++++-------
.../controller/PartitionLeaderSelector.scala | 40 ++--
.../controller/PartitionStateMachine.scala | 22 +-
.../kafka/controller/ReplicaStateMachine.scala | 35 ++--
.../kafka/javaapi/TopicMetadataRequest.scala | 25 +++
core/src/main/scala/kafka/log/Log.scala | 37 ++--
core/src/main/scala/kafka/log/LogCleaner.scala | 208 +++++++------------
.../scala/kafka/log/LogCleanerManager.scala | 188 +++++++++++++++++
core/src/main/scala/kafka/log/LogManager.scala | 76 ++++---
.../scala/kafka/network/RequestChannel.scala | 6 +-
.../scala/kafka/producer/SyncProducer.scala | 5 -
.../kafka/server/AbstractFetcherThread.scala | 10 +-
.../src/main/scala/kafka/server/KafkaApis.scala | 5 +-
.../kafka/server/KafkaRequestHandler.scala | 1 +
.../kafka/server/ReplicaFetcherThread.scala | 7 +-
.../scala/kafka/server/ReplicaManager.scala | 18 +-
.../kafka/server/ZookeeperLeaderElector.scala | 36 ++--
core/src/main/scala/kafka/utils/ZkUtils.scala | 43 ----
.../test/scala/unit/kafka/admin/AdminTest.scala | 15 +-
.../test/scala/unit/kafka/log/CleanerTest.scala | 39 ++--
examples/README | 6 +-
system_test/utils/kafka_system_test_utils.py | 6 +-
49 files changed, 913 insertions(+), 513 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/3f473566/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/kafka/controller/KafkaController.scala
index ca2f09b,a0267ae..f4f00b2
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@@ -488,7 -546,7 +546,7 @@@ class KafkaController(val config : Kafk
}
}
-- def onPreferredReplicaElection(partitions: Set[TopicAndPartition]) {
++ def onPreferredReplicaElection(partitions: Set[TopicAndPartition], isTriggeredByAutoRebalance: Boolean = true) {
info("Starting preferred replica leader election for partitions %s".format(partitions.mkString(",")))
try {
controllerContext.partitionsUndergoingPreferredReplicaElection ++= partitions
@@@ -496,7 -554,7 +554,7 @@@
} catch {
case e: Throwable => error("Error completing preferred replica leader election for partitions %s".format(partitions.mkString(",")), e)
} finally {
-- removePartitionsFromPreferredReplicaElection(partitions)
++ removePartitionsFromPreferredReplicaElection(partitions, isTriggeredByAutoRebalance)
}
}
@@@ -765,7 -823,7 +823,8 @@@
}
}
-- def removePartitionsFromPreferredReplicaElection(partitionsToBeRemoved: Set[TopicAndPartition]) {
++ def removePartitionsFromPreferredReplicaElection(partitionsToBeRemoved: Set[TopicAndPartition],
++ isTriggeredByAutoRebalance : Boolean) {
for(partition <- partitionsToBeRemoved) {
// check the status
val currentLeader = controllerContext.partitionLeadershipInfo(partition).leaderAndIsr.leader
@@@ -776,7 -834,7 +835,8 @@@
warn("Partition %s failed to complete preferred replica leader election. Leader is %d".format(partition, currentLeader))
}
}
-- ZkUtils.deletePath(zkClient, ZkUtils.PreferredReplicaLeaderElectionPath)
++ if (!isTriggeredByAutoRebalance)
++ ZkUtils.deletePath(zkClient, ZkUtils.PreferredReplicaLeaderElectionPath)
controllerContext.partitionsUndergoingPreferredReplicaElection --= partitionsToBeRemoved
}
@@@ -966,26 -1018,26 +1020,16 @@@
// check ratio and if greater than desired ratio, trigger a rebalance for the topic partitions
// that need to be on this broker
if (imbalanceRatio > (config.leaderImbalancePerBrokerPercentage.toDouble / 100)) {
-- controllerContext.controllerLock synchronized {
-- // do this check only if the broker is live and there are no partitions being reassigned currently
-- // and preferred replica election is not in progress
-- if (controllerContext.liveBrokerIds.contains(leaderBroker) &&
-- controllerContext.partitionsBeingReassigned.size == 0 &&
-- controllerContext.partitionsUndergoingPreferredReplicaElection.size == 0) {
-- val zkPath = ZkUtils.PreferredReplicaLeaderElectionPath
-- val partitionsList = topicsNotInPreferredReplica.keys.map(e => Map("topic" -> e.topic, "partition" -> e.partition))
-- val jsonData = Json.encode(Map("version" -> 1, "partitions" -> partitionsList))
-- try {
-- ZkUtils.createPersistentPath(zkClient, zkPath, jsonData)
-- info("Created preferred replica election path with %s".format(jsonData))
-- } catch {
-- case e2: ZkNodeExistsException =>
-- val partitionsUndergoingPreferredReplicaElection =
-- PreferredReplicaLeaderElectionCommand.parsePreferredReplicaElectionData(ZkUtils.readData(zkClient, zkPath)._1)
-- error("Preferred replica leader election currently in progress for " +
-- "%s. Aborting operation".format(partitionsUndergoingPreferredReplicaElection));
-- case e3: Throwable =>
-- error("Error while trying to auto rebalance topics %s".format(topicsNotInPreferredReplica.keys))
++ topicsNotInPreferredReplica.foreach {
++ case(topicPartition, replicas) => {
++ controllerContext.controllerLock synchronized {
++ // do this check only if the broker is live and there are no partitions being reassigned currently
++ // and preferred replica election is not in progress
++ if (controllerContext.liveBrokerIds.contains(leaderBroker) &&
++ controllerContext.partitionsBeingReassigned.size == 0 &&
++ controllerContext.partitionsUndergoingPreferredReplicaElection.size == 0) {
++ onPreferredReplicaElection(Set(topicPartition), false)
++ }
}
}
}
[14/19] git commit: Merge branch 'trunk' of
http://git-wip-us.apache.org/repos/asf/kafka into trunk
Posted by sr...@apache.org.
Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into trunk
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/72610d1f
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/72610d1f
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/72610d1f
Branch: refs/heads/trunk
Commit: 72610d1fc0bbf706a14f1605bf9862ea3830e707
Parents: e88f1ac b5d1687
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Fri Dec 20 13:01:45 2013 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Fri Dec 20 13:01:45 2013 -0800
----------------------------------------------------------------------
config/server.properties | 5 +-
.../main/java/kafka/etl/impl/DataGenerator.java | 3 +-
.../kafka/admin/ReassignPartitionsCommand.scala | 231 ++++++++++---------
.../main/scala/kafka/admin/TopicCommand.scala | 133 ++++++-----
.../src/main/scala/kafka/api/FetchRequest.scala | 2 +-
.../scala/kafka/api/RequestOrResponse.scala | 3 +
.../main/scala/kafka/cluster/Partition.scala | 12 +-
.../scala/kafka/consumer/ConsumerIterator.scala | 5 +-
.../main/scala/kafka/consumer/TopicFilter.scala | 6 -
.../consumer/ZookeeperConsumerConnector.scala | 37 ++-
.../consumer/ZookeeperTopicEventWatcher.scala | 19 +-
.../controller/ControllerChannelManager.scala | 11 +-
.../kafka/controller/KafkaController.scala | 32 ++-
.../controller/PartitionStateMachine.scala | 3 +-
.../kafka/controller/ReplicaStateMachine.scala | 29 +--
.../main/scala/kafka/log/FileMessageSet.scala | 2 +
core/src/main/scala/kafka/log/Log.scala | 28 ++-
core/src/main/scala/kafka/log/LogManager.scala | 2 +
core/src/main/scala/kafka/log/OffsetIndex.scala | 40 ++--
.../kafka/message/MessageAndMetadata.scala | 15 +-
.../kafka/producer/KafkaLog4jAppender.scala | 12 +-
.../kafka/server/AbstractFetcherManager.scala | 4 +-
.../src/main/scala/kafka/server/KafkaApis.scala | 13 +-
.../main/scala/kafka/server/KafkaConfig.scala | 17 +-
.../scala/kafka/server/ReplicaManager.scala | 169 ++++++++------
.../kafka/tools/ReplicaVerificationTool.scala | 60 ++---
.../test/scala/unit/kafka/admin/AdminTest.scala | 45 +++-
.../kafka/consumer/ConsumerIteratorTest.scala | 36 +++
.../unit/kafka/consumer/TopicFilterTest.scala | 4 -
.../scala/unit/kafka/log/LogSegmentTest.scala | 5 +-
.../src/test/scala/unit/kafka/log/LogTest.scala | 22 +-
.../unit/kafka/server/ISRExpirationTest.scala | 1 +
.../unit/kafka/server/LogRecoveryTest.scala | 1 +
.../unit/kafka/server/SimpleFetchTest.scala | 1 +
.../test/scala/unit/kafka/utils/TestUtils.scala | 10 +-
kafka-patch-review.py | 15 ++
project/Build.scala | 4 +-
.../config/server.properties | 5 +-
.../config/server.properties | 5 +-
.../config/server.properties | 5 +-
40 files changed, 605 insertions(+), 447 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/72610d1f/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/72610d1f/core/src/main/scala/kafka/server/KafkaConfig.scala
----------------------------------------------------------------------
[02/19] git commit: commit the remaining changes
Posted by sr...@apache.org.
commit the remaining changes
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/b998f5d8
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/b998f5d8
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/b998f5d8
Branch: refs/heads/trunk
Commit: b998f5d8790cf6f4d079cba399f5c5ea8ea43e68
Parents: c304a92
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Wed Oct 9 10:47:18 2013 -0700
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Wed Oct 9 10:47:18 2013 -0700
----------------------------------------------------------------------
core/src/main/scala/kafka/admin/AdminUtils.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/b998f5d8/core/src/main/scala/kafka/admin/AdminUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/AdminUtils.scala b/core/src/main/scala/kafka/admin/AdminUtils.scala
index 1360955..9a8b0c9 100644
--- a/core/src/main/scala/kafka/admin/AdminUtils.scala
+++ b/core/src/main/scala/kafka/admin/AdminUtils.scala
@@ -100,7 +100,7 @@ object AdminUtils extends Logging {
// create the new partition replication list
val brokerList = ZkUtils.getSortedBrokerList(zkClient)
- val newPartitionReplicaList = if (replicaAssignmentStr == "")
+ val newPartitionReplicaList = if (replicaAssignmentStr == null || replicaAssignmentStr == "")
AdminUtils.assignReplicasToBrokers(brokerList, partitionsToAdd, existingReplicaList.size, existingReplicaList.head, existingPartitionsReplicaList.size)
else
getManualReplicaAssignment(replicaAssignmentStr, brokerList.toSet, existingPartitionsReplicaList.size)
[11/19] git commit: Address code review feedbacks
Posted by sr...@apache.org.
Address code review feedbacks
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/55d77c67
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/55d77c67
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/55d77c67
Branch: refs/heads/trunk
Commit: 55d77c67c236488df2b9f6bb59d99eeb645a0553
Parents: 7f32a1c
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Mon Dec 9 22:51:41 2013 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Mon Dec 9 22:51:41 2013 -0800
----------------------------------------------------------------------
.../kafka/controller/KafkaController.scala | 87 +++++++++++++-------
1 file changed, 56 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/55d77c67/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index 6cd58a4..522e6c7 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -143,6 +143,22 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
}
)
+ newGauge(
+ "PreferredReplicaImbalanceCount",
+ new Gauge[Int] {
+ def value(): Int = {
+ controllerContext.controllerLock synchronized {
+ if (!isActive())
+ 0
+ else
+ controllerContext.partitionReplicaAssignment.count {
+ case (topicPartition, replicas) => controllerContext.partitionLeadershipInfo(topicPartition).leaderAndIsr.leader != replicas.head
+ }
+ }
+ }
+ }
+ )
+
def epoch = controllerContext.epoch
def clientId = "id_%d-host_%s-port_%d".format(config.brokerId, config.hostName, config.port)
@@ -465,7 +481,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
}
}
- def onPreferredReplicaElection(partitions: Set[TopicAndPartition], updateZk: Boolean = true) {
+ def onPreferredReplicaElection(partitions: Set[TopicAndPartition], isTriggeredByAutoRebalance: Boolean = true) {
info("Starting preferred replica leader election for partitions %s".format(partitions.mkString(",")))
try {
controllerContext.partitionsUndergoingPreferredReplicaElection ++= partitions
@@ -473,7 +489,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
} catch {
case e: Throwable => error("Error completing preferred replica leader election for partitions %s".format(partitions.mkString(",")), e)
} finally {
- removePartitionsFromPreferredReplicaElection(partitions, updateZk)
+ removePartitionsFromPreferredReplicaElection(partitions, isTriggeredByAutoRebalance)
}
}
@@ -742,7 +758,8 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
}
}
- def removePartitionsFromPreferredReplicaElection(partitionsToBeRemoved: Set[TopicAndPartition], updateZK : Boolean) {
+ def removePartitionsFromPreferredReplicaElection(partitionsToBeRemoved: Set[TopicAndPartition],
+ isTriggeredByAutoRebalance : Boolean) {
for(partition <- partitionsToBeRemoved) {
// check the status
val currentLeader = controllerContext.partitionLeadershipInfo(partition).leaderAndIsr.leader
@@ -753,7 +770,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
warn("Partition %s failed to complete preferred replica leader election. Leader is %d".format(partition, currentLeader))
}
}
- if (updateZK)
+ if (isTriggeredByAutoRebalance)
ZkUtils.deletePath(zkClient, ZkUtils.PreferredReplicaLeaderElectionPath)
controllerContext.partitionsUndergoingPreferredReplicaElection --= partitionsToBeRemoved
}
@@ -913,42 +930,50 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
private def checkAndTriggerPartitionRebalance(): Unit = {
if (isActive()) {
- info("checking need to trigger partition rebalance")
+ trace("checking need to trigger partition rebalance")
// get all the active brokers
- var preferredReplicasForTopicsByBrokers: Map[Int, Map[TopicAndPartition, Seq[Int]]] = null;
+ var preferredReplicasForTopicsByBrokers: Map[Int, Map[TopicAndPartition, Seq[Int]]] = null
controllerContext.controllerLock synchronized {
- preferredReplicasForTopicsByBrokers = controllerContext.partitionReplicaAssignment.groupBy(_._2.head)
+ preferredReplicasForTopicsByBrokers = controllerContext.partitionReplicaAssignment.groupBy {
+ case(topicAndPartition, assignedReplicas) => assignedReplicas.head
+ }
}
debug("preferred replicas by broker " + preferredReplicasForTopicsByBrokers)
// for each broker, check if a preferred replica election needs to be triggered
- preferredReplicasForTopicsByBrokers.foreach( brokerInfo => {
- var imbalanceRatio: Double = 0
- var topicsNotInPreferredReplica: Map[TopicAndPartition, Seq[Int]] = null
- controllerContext.controllerLock synchronized {
- val brokerIds = controllerContext.liveBrokerIds
- if (brokerIds.contains(brokerInfo._1) &&
- controllerContext.partitionsBeingReassigned.size == 0) {
- // do this check only if the broker is live and there are no partitions being reassigned currently
- topicsNotInPreferredReplica =
- brokerInfo._2.filter(item => controllerContext.partitionLeadershipInfo(item._1).leaderAndIsr.leader != brokerInfo._1);
- debug("topics not in preferred replica " + topicsNotInPreferredReplica)
- val totalTopicPartitionsForBroker = brokerInfo._2.size
- val totalTopicPartitionsNotLedByBroker = topicsNotInPreferredReplica.size
- imbalanceRatio = totalTopicPartitionsNotLedByBroker.toDouble / totalTopicPartitionsForBroker
- info("leader imbalance ratio for broker %d is %f".format(brokerInfo._1, imbalanceRatio))
+ preferredReplicasForTopicsByBrokers.foreach {
+ case(leaderBroker, topicAndPartitionsForBroker) => {
+ var imbalanceRatio: Double = 0
+ var topicsNotInPreferredReplica: Map[TopicAndPartition, Seq[Int]] = null
+ controllerContext.controllerLock synchronized {
+ if (controllerContext.liveBrokerIds.contains(leaderBroker) &&
+ controllerContext.partitionsBeingReassigned.size == 0) {
+ // do this check only if the broker is live and there are no partitions being reassigned currently
+ topicsNotInPreferredReplica =
+ topicAndPartitionsForBroker.filter {
+ case(topicPartition, replicas) => {
+ controllerContext.partitionLeadershipInfo(topicPartition).leaderAndIsr.leader != leaderBroker
+ }
+ }
+ debug("topics not in preferred replica " + topicsNotInPreferredReplica)
+ val totalTopicPartitionsForBroker = topicAndPartitionsForBroker.size
+ val totalTopicPartitionsNotLedByBroker = topicsNotInPreferredReplica.size
+ imbalanceRatio = totalTopicPartitionsNotLedByBroker.toDouble / totalTopicPartitionsForBroker
+ trace("leader imbalance ratio for broker %d is %f".format(leaderBroker, imbalanceRatio))
+ }
}
- }
- // check ratio and if greater than desired ratio, trigger a rebalance for the topics
- // that need to be on this broker
- if (imbalanceRatio > (config.leaderImbalancePerBrokerPercentage.toDouble / 100)) {
- topicsNotInPreferredReplica.foreach(topicPartition => {
- controllerContext.controllerLock synchronized {
- onPreferredReplicaElection(Set(topicPartition._1), false)
+ // check ratio and if greater than desired ratio, trigger a rebalance for the topic partitions
+ // that need to be on this broker
+ if (imbalanceRatio > (config.leaderImbalancePerBrokerPercentage.toDouble / 100)) {
+ topicsNotInPreferredReplica.foreach {
+ case(topicPartition, replicas) => {
+ controllerContext.controllerLock synchronized {
+ onPreferredReplicaElection(Set(topicPartition), false)
+ }
+ }
}
- })
+ }
}
}
- )
}
}
}
[12/19] git commit: use zk for auto rebalance
Posted by sr...@apache.org.
use zk for auto rebalance
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/425af9b4
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/425af9b4
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/425af9b4
Branch: refs/heads/trunk
Commit: 425af9b4a7afeb21191c33ba6bc3f20623a3f0b3
Parents: 55d77c6
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Fri Dec 20 11:11:20 2013 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Fri Dec 20 11:11:20 2013 -0800
----------------------------------------------------------------------
.../kafka/controller/KafkaController.scala | 48 ++++++++++++--------
1 file changed, 29 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/425af9b4/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index 522e6c7..74e2ea4 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -21,7 +21,7 @@ import collection.immutable.Set
import com.yammer.metrics.core.Gauge
import java.lang.{IllegalStateException, Object}
import java.util.concurrent.TimeUnit
-import kafka.admin.PreferredReplicaLeaderElectionCommand
+import kafka.admin.{AdminOperationException, PreferredReplicaLeaderElectionCommand}
import kafka.api._
import kafka.cluster.Broker
import kafka.common._
@@ -945,29 +945,39 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
var imbalanceRatio: Double = 0
var topicsNotInPreferredReplica: Map[TopicAndPartition, Seq[Int]] = null
controllerContext.controllerLock synchronized {
- if (controllerContext.liveBrokerIds.contains(leaderBroker) &&
- controllerContext.partitionsBeingReassigned.size == 0) {
- // do this check only if the broker is live and there are no partitions being reassigned currently
- topicsNotInPreferredReplica =
- topicAndPartitionsForBroker.filter {
- case(topicPartition, replicas) => {
- controllerContext.partitionLeadershipInfo(topicPartition).leaderAndIsr.leader != leaderBroker
- }
+ topicsNotInPreferredReplica =
+ topicAndPartitionsForBroker.filter {
+ case(topicPartition, replicas) => {
+ controllerContext.partitionLeadershipInfo(topicPartition).leaderAndIsr.leader != leaderBroker
}
- debug("topics not in preferred replica " + topicsNotInPreferredReplica)
- val totalTopicPartitionsForBroker = topicAndPartitionsForBroker.size
- val totalTopicPartitionsNotLedByBroker = topicsNotInPreferredReplica.size
- imbalanceRatio = totalTopicPartitionsNotLedByBroker.toDouble / totalTopicPartitionsForBroker
- trace("leader imbalance ratio for broker %d is %f".format(leaderBroker, imbalanceRatio))
- }
+ }
+ debug("topics not in preferred replica " + topicsNotInPreferredReplica)
+ val totalTopicPartitionsForBroker = topicAndPartitionsForBroker.size
+ val totalTopicPartitionsNotLedByBroker = topicsNotInPreferredReplica.size
+ imbalanceRatio = totalTopicPartitionsNotLedByBroker.toDouble / totalTopicPartitionsForBroker
+ trace("leader imbalance ratio for broker %d is %f".format(leaderBroker, imbalanceRatio))
}
// check ratio and if greater than desired ratio, trigger a rebalance for the topic partitions
// that need to be on this broker
if (imbalanceRatio > (config.leaderImbalancePerBrokerPercentage.toDouble / 100)) {
- topicsNotInPreferredReplica.foreach {
- case(topicPartition, replicas) => {
- controllerContext.controllerLock synchronized {
- onPreferredReplicaElection(Set(topicPartition), false)
+ controllerContext.controllerLock synchronized {
+ // do this check only if the broker is live and there are no partitions being reassigned currently
+ // and preferred replica election is not in progress
+ if (controllerContext.liveBrokerIds.contains(leaderBroker) &&
+ controllerContext.partitionsBeingReassigned.size == 0 &&
+ controllerContext.partitionsUndergoingPreferredReplicaElection.size == 0) {
+ val zkPath = ZkUtils.PreferredReplicaLeaderElectionPath
+ val partitionsList = topicsNotInPreferredReplica.keys.map(e => Map("topic" -> e.topic, "partition" -> e.partition))
+ val jsonData = Json.encode(Map("version" -> 1, "partitions" -> partitionsList))
+ try {
+ ZkUtils.createPersistentPath(zkClient, zkPath, jsonData)
+ info("Created preferred replica election path with %s".format(jsonData))
+ } catch {
+ case e2: Throwable =>
+ val partitionsUndergoingPreferredReplicaElection =
+ PreferredReplicaLeaderElectionCommand.parsePreferredReplicaElectionData(ZkUtils.readData(zkClient, zkPath)._1)
+ error("Preferred replica leader election currently in progress for " +
+ "%s. Aborting operation".format(partitionsUndergoingPreferredReplicaElection));
}
}
}
[18/19] git commit: Merge branch 'trunk' of
http://git-wip-us.apache.org/repos/asf/kafka into trunk
Posted by sr...@apache.org.
Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into trunk
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/5b80758b
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/5b80758b
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/5b80758b
Branch: refs/heads/trunk
Commit: 5b80758b57d8063b8a8124562cc5c83854a99e00
Parents: 5de68ef 3955915
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Tue Feb 25 00:22:04 2014 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Tue Feb 25 00:22:04 2014 -0800
----------------------------------------------------------------------
.../main/scala/kafka/network/SocketServer.scala | 16 +++++++--
.../unit/kafka/network/SocketServerTest.scala | 38 +++++++++-----------
2 files changed, 31 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
[03/19] git commit: change comments
Posted by sr...@apache.org.
change comments
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/7c920e97
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/7c920e97
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/7c920e97
Branch: refs/heads/trunk
Commit: 7c920e97ef9e2db80774d6ccd9b85e3934e92645
Parents: b998f5d
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Wed Oct 9 10:54:09 2013 -0700
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Wed Oct 9 10:54:09 2013 -0700
----------------------------------------------------------------------
core/src/main/scala/kafka/admin/TopicCommand.scala | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/7c920e97/core/src/main/scala/kafka/admin/TopicCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala
index ee940e9..56f3177 100644
--- a/core/src/main/scala/kafka/admin/TopicCommand.scala
+++ b/core/src/main/scala/kafka/admin/TopicCommand.scala
@@ -85,7 +85,8 @@ object TopicCommand {
println("Updated config for topic \"%s\".".format(topic))
}
if(opts.options.has(opts.partitionsOpt)) {
- println("partitions can only be added when topic has no key")
+ println("WARNING: If partitions are increased for a topic that has a key, the partition " +
+ "logic or ordering of the messages will be affected")
val nPartitions = opts.options.valueOf(opts.partitionsOpt).intValue
val replicaAssignmentStr = opts.options.valueOf(opts.replicaAssignmentOpt)
AdminUtils.addPartitions(zkClient, topic, nPartitions, replicaAssignmentStr)
@@ -188,7 +189,7 @@ object TopicCommand {
.describedAs("name=value")
.ofType(classOf[String])
val partitionsOpt = parser.accepts("partitions", "The number of partitions for the topic being created or " +
- "altered (Partitions can only be added for a topic which has no key. Partitions cannot be decreased")
+ "altered (WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected")
.withRequiredArg
.describedAs("# of partitions")
.ofType(classOf[java.lang.Integer])
[19/19] git commit: Check delete state of topic
Posted by sr...@apache.org.
Check delete state of topic
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/00afb619
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/00afb619
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/00afb619
Branch: refs/heads/trunk
Commit: 00afb619cb1f2aaca0ddc785ab1d649428f9c93e
Parents: 5b80758
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Tue Feb 25 00:26:14 2014 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Tue Feb 25 00:26:14 2014 -0800
----------------------------------------------------------------------
core/src/main/scala/kafka/controller/KafkaController.scala | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/00afb619/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index f12ffc2..b58cdcd 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -1114,7 +1114,6 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
controllerContext.partitionsBeingReassigned.size == 0 &&
controllerContext.partitionsUndergoingPreferredReplicaElection.size == 0 &&
!deleteTopicManager.isTopicQueuedUpForDeletion(topicPartition.topic) &&
- !deleteTopicManager.isTopicDeletionInProgress(topicPartition.topic) &&
controllerContext.allTopics.contains(topicPartition.topic)) {
onPreferredReplicaElection(Set(topicPartition), false)
}
[13/19] git commit: some more changes
Posted by sr...@apache.org.
some more changes
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/e88f1acd
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/e88f1acd
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/e88f1acd
Branch: refs/heads/trunk
Commit: e88f1acdaac9fe738787ccb8375519293c913a37
Parents: 425af9b
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Fri Dec 20 11:22:26 2013 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Fri Dec 20 11:22:26 2013 -0800
----------------------------------------------------------------------
.../main/scala/kafka/controller/KafkaController.scala | 14 +++++++-------
1 file changed, 7 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/e88f1acd/core/src/main/scala/kafka/controller/KafkaController.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala
index 74e2ea4..8017abb 100644
--- a/core/src/main/scala/kafka/controller/KafkaController.scala
+++ b/core/src/main/scala/kafka/controller/KafkaController.scala
@@ -481,7 +481,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
}
}
- def onPreferredReplicaElection(partitions: Set[TopicAndPartition], isTriggeredByAutoRebalance: Boolean = true) {
+ def onPreferredReplicaElection(partitions: Set[TopicAndPartition]) {
info("Starting preferred replica leader election for partitions %s".format(partitions.mkString(",")))
try {
controllerContext.partitionsUndergoingPreferredReplicaElection ++= partitions
@@ -489,7 +489,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
} catch {
case e: Throwable => error("Error completing preferred replica leader election for partitions %s".format(partitions.mkString(",")), e)
} finally {
- removePartitionsFromPreferredReplicaElection(partitions, isTriggeredByAutoRebalance)
+ removePartitionsFromPreferredReplicaElection(partitions)
}
}
@@ -758,8 +758,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
}
}
- def removePartitionsFromPreferredReplicaElection(partitionsToBeRemoved: Set[TopicAndPartition],
- isTriggeredByAutoRebalance : Boolean) {
+ def removePartitionsFromPreferredReplicaElection(partitionsToBeRemoved: Set[TopicAndPartition]) {
for(partition <- partitionsToBeRemoved) {
// check the status
val currentLeader = controllerContext.partitionLeadershipInfo(partition).leaderAndIsr.leader
@@ -770,8 +769,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
warn("Partition %s failed to complete preferred replica leader election. Leader is %d".format(partition, currentLeader))
}
}
- if (isTriggeredByAutoRebalance)
- ZkUtils.deletePath(zkClient, ZkUtils.PreferredReplicaLeaderElectionPath)
+ ZkUtils.deletePath(zkClient, ZkUtils.PreferredReplicaLeaderElectionPath)
controllerContext.partitionsUndergoingPreferredReplicaElection --= partitionsToBeRemoved
}
@@ -973,11 +971,13 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient) extends Logg
ZkUtils.createPersistentPath(zkClient, zkPath, jsonData)
info("Created preferred replica election path with %s".format(jsonData))
} catch {
- case e2: Throwable =>
+ case e2: ZkNodeExistsException =>
val partitionsUndergoingPreferredReplicaElection =
PreferredReplicaLeaderElectionCommand.parsePreferredReplicaElectionData(ZkUtils.readData(zkClient, zkPath)._1)
error("Preferred replica leader election currently in progress for " +
"%s. Aborting operation".format(partitionsUndergoingPreferredReplicaElection));
+ case e3: Throwable =>
+ error("Error while trying to auto rebalance topics %s".format(topicsNotInPreferredReplica.keys))
}
}
}
[05/19] git commit: Merge branch 'trunk' of
http://git-wip-us.apache.org/repos/asf/kafka into trunk
Posted by sr...@apache.org.
Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into trunk
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/39c5b75c
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/39c5b75c
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/39c5b75c
Branch: refs/heads/trunk
Commit: 39c5b75c9675636d1f77a549fc55b527ed1d7f38
Parents: 793b607 58789d7
Author: Sriram Subramanian <sr...@gmail.com>
Authored: Tue Nov 19 16:04:16 2013 -0800
Committer: Sriram Subramanian <sr...@gmail.com>
Committed: Tue Nov 19 16:04:16 2013 -0800
----------------------------------------------------------------------
bin/kafka-run-class.sh | 43 +++++++++++++++++---
bin/kafka-server-start.sh | 17 +++++++-
bin/zookeeper-server-start.sh | 17 +++++++-
config/log4j.properties | 11 +++--
.../main/scala/kafka/log/FileMessageSet.scala | 2 +
core/src/main/scala/kafka/log/Log.scala | 28 ++++++-------
core/src/main/scala/kafka/log/LogManager.scala | 2 +
core/src/main/scala/kafka/log/OffsetIndex.scala | 40 ++++++++++--------
.../scala/unit/kafka/log/LogSegmentTest.scala | 5 +--
.../src/test/scala/unit/kafka/log/LogTest.scala | 22 +++++-----
.../test/scala/unit/kafka/utils/TestUtils.scala | 10 ++++-
11 files changed, 136 insertions(+), 61 deletions(-)
----------------------------------------------------------------------