diff --git a/core/src/main/scala/kafka/common/InconsistentBrokerIdException.scala b/core/src/main/scala/kafka/common/InconsistentBrokerIdException.scala new file mode 100644 index 0000000..5649828 --- /dev/null +++ b/core/src/main/scala/kafka/common/InconsistentBrokerIdException.scala @@ -0,0 +1,25 @@ +/** + * 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.common + +/** + * Indicates the client has requested a range no longer available on the server + */ +class InconsistentBrokerIdException(message: String) extends RuntimeException(message) { + def this() = this(null) +} diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala b/core/src/main/scala/kafka/log/LogCleaner.scala index afbeffc..2faa196 100644 --- a/core/src/main/scala/kafka/log/LogCleaner.scala +++ b/core/src/main/scala/kafka/log/LogCleaner.scala @@ -28,8 +28,6 @@ import kafka.utils._ import kafka.metrics.KafkaMetricsGroup import com.yammer.metrics.core.Gauge import java.lang.IllegalStateException -import java.util.concurrent.CountDownLatch -import java.util.concurrent.TimeUnit /** * The cleaner is responsible for removing obsolete records from logs which have the dedupe retention strategy. @@ -175,8 +173,7 @@ class LogCleaner(val config: CleanerConfig, checkDone = checkDone) @volatile var lastStats: CleanerStats = new CleanerStats() - private val backOffWaitLatch = new CountDownLatch(1) - + private def checkDone(topicAndPartition: TopicAndPartition) { if (!isRunning.get()) throw new ThreadShutdownException @@ -190,13 +187,6 @@ class LogCleaner(val config: CleanerConfig, cleanOrSleep() } - - override def shutdown() = { - initiateShutdown() - backOffWaitLatch.countDown() - awaitShutdown() - } - /** * Clean a log if there is a dirty log available, otherwise sleep for a bit */ @@ -204,7 +194,7 @@ class LogCleaner(val config: CleanerConfig, cleanerManager.grabFilthiestLog() match { case None => // there are no cleanable logs, sleep a while - backOffWaitLatch.await(config.backOffMs, TimeUnit.MILLISECONDS) + time.sleep(config.backOffMs) case Some(cleanable) => // there's a log, clean it var endOffset = cleanable.firstDirtyOffset diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index 50b09ed..c65d60c 100644 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -35,13 +35,13 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro private def getLogRetentionTimeMillis(): Long = { val millisInMinute = 60L * 1000L val millisInHour = 60L * millisInMinute - + if(props.containsKey("log.retention.ms")){ props.getIntInRange("log.retention.ms", (1, Int.MaxValue)) } else if(props.containsKey("log.retention.minutes")){ millisInMinute * props.getIntInRange("log.retention.minutes", (1, Int.MaxValue)) - } + } else { millisInHour * props.getIntInRange("log.retention.hours", 24*7, (1, Int.MaxValue)) } @@ -49,7 +49,7 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro private def getLogRollTimeMillis(): Long = { val millisInHour = 60L * 60L * 1000L - + if(props.containsKey("log.roll.ms")){ props.getIntInRange("log.roll.ms", (1, Int.MaxValue)) } @@ -57,11 +57,17 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro millisInHour * props.getIntInRange("log.roll.hours", 24*7, (1, Int.MaxValue)) } } - + /*********** General Configuration ***********/ - /* the broker id for this server */ - val brokerId: Int = props.getIntInRange("broker.id", (0, Int.MaxValue)) + /* Max number that can be used for a broker.id */ + val MaxReservedBrokerId = props.getIntInRange("reserved.broker.max.id", 1000, (0, Int.MaxValue)) + + /* The broker id for this server. + * To avoid conflicts between zookeeper generated brokerId and user's config.brokerId + * added MaxReservedBrokerId and zookeeper sequence starts from MaxReservedBrokerId + 1. + */ + var brokerId: Int = if (props.containsKey("broker.id")) props.getIntInRange("broker.id", (0, MaxReservedBrokerId)) else -1 /* the maximum size of message that the server can receive */ val messageMaxBytes = props.getIntInRange("message.max.bytes", 1000000 + MessageSet.LogOverhead, (0, Int.MaxValue)) @@ -106,10 +112,10 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro /* the maximum number of bytes in a socket request */ val socketRequestMaxBytes: Int = props.getIntInRange("socket.request.max.bytes", 100*1024*1024, (1, Int.MaxValue)) - + /* the maximum number of connections we allow from each ip address */ val maxConnectionsPerIp: Int = props.getIntInRange("max.connections.per.ip", Int.MaxValue, (1, Int.MaxValue)) - + /* per-ip or hostname overrides to the default maximum number of connections */ val maxConnectionsPerIpOverrides = props.getMap("max.connections.per.ip.overrides").map(entry => (entry._1, entry._2.toInt)) diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index def1dc2..d70496d 100644 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -25,11 +25,16 @@ import kafka.utils._ import java.util.concurrent._ import atomic.{AtomicInteger, AtomicBoolean} import java.io.File +import java.io.FileOutputStream +import java.io.FileNotFoundException +import java.util.Properties +import collection.mutable import org.I0Itec.zkclient.ZkClient import kafka.controller.{ControllerStats, KafkaController} import kafka.cluster.Broker import kafka.api.{ControlledShutdownResponse, ControlledShutdownRequest} import kafka.common.ErrorMapping +import kafka.common.InconsistentBrokerIdException import kafka.network.{Receive, BlockingChannel, SocketServer} import kafka.metrics.KafkaMetricsGroup import com.yammer.metrics.core.Gauge @@ -39,10 +44,12 @@ import com.yammer.metrics.core.Gauge * to start up and shutdown a single Kafka node. */ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime) extends Logging with KafkaMetricsGroup { - this.logIdent = "[Kafka Server " + config.brokerId + "], " + private var isShuttingDown = new AtomicBoolean(false) private var shutdownLatch = new CountDownLatch(1) private var startupComplete = new AtomicBoolean(false) + private var brokerId: Int = -1 + private var metaPropsFile = "meta.properties" val brokerState: BrokerState = new BrokerState val correlationId: AtomicInteger = new AtomicInteger(0) var socketServer: SocketServer = null @@ -76,7 +83,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime) extends Logg /* start scheduler */ kafkaScheduler.startup() - + /* setup zookeeper */ zkClient = initZk() @@ -84,6 +91,10 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime) extends Logg logManager = createLogManager(zkClient, brokerState) logManager.startup() + /* generate brokerId */ + config.brokerId = getBrokerId(zkClient, config) + this.logIdent = "[Kafka Server " + config.brokerId + "], " + socketServer = new SocketServer(config.brokerId, config.hostName, config.port, @@ -101,31 +112,31 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime) extends Logg offsetManager = createOffsetManager() kafkaController = new KafkaController(config, zkClient, brokerState) - + /* start processing requests */ apis = new KafkaApis(socketServer.requestChannel, replicaManager, offsetManager, zkClient, config.brokerId, config, kafkaController) requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.requestChannel, apis, config.numIoThreads) brokerState.newState(RunningAsBroker) - + Mx4jLoader.maybeLoad() replicaManager.startup() kafkaController.startup() - + topicConfigManager = new TopicConfigManager(zkClient, logManager) topicConfigManager.startup() - + /* tell everyone we are alive */ kafkaHealthcheck = new KafkaHealthcheck(config.brokerId, config.advertisedHostName, config.advertisedPort, config.zkSessionTimeoutMs, zkClient) kafkaHealthcheck.startup() - + registerStats() startupComplete.set(true) info("started") } - + private def initZk(): ZkClient = { info("Connecting to zookeeper on " + config.zkConnect) val zkClient = new ZkClient(config.zkConnect, config.zkSessionTimeoutMs, config.zkConnectionTimeoutMs, ZKStringSerializer) @@ -273,9 +284,9 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime) extends Logg def awaitShutdown(): Unit = shutdownLatch.await() def getLogManager(): LogManager = logManager - + private def createLogManager(zkClient: ZkClient, brokerState: BrokerState): LogManager = { - val defaultLogConfig = LogConfig(segmentSize = config.logSegmentBytes, + val defaultLogConfig = LogConfig(segmentSize = config.logSegmentBytes, segmentMs = config.logRollTimeMillis, flushInterval = config.logFlushIntervalMessages, flushMs = config.logFlushIntervalMs.toLong, @@ -323,5 +334,73 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime) extends Logg new OffsetManager(offsetManagerConfig, replicaManager, zkClient, kafkaScheduler) } -} + /** + * if kafka server config has brokerId and there is no meta.properties file returns the config.brokerId, + * otherwise generates a sequence id from ZK uses it as the brokerId. + * stores the generated zk sequence id in meta.properties under logDirs specified in config. + * if config has brokerId and meta.properties contains brokerId if they don't match throws InconsistentBrokerIdException + */ + private def getBrokerId(zkClient: ZkClient, config: KafkaConfig): Int = { + var brokerId = config.brokerId + var logDirsWithoutMetaProps: List[String] = List() + val metaBrokerIdSet = mutable.HashSet[Int]() + + for (logDir <- config.logDirs) { + val metaBrokerIdOpt = readBrokerIdFromMetaProps(logDir) + metaBrokerIdOpt match { + case Some(metaBrokerId) => + metaBrokerIdSet.add(metaBrokerId) + case None => + logDirsWithoutMetaProps ++= List(logDir) + } + } + + if(metaBrokerIdSet.size > 1) { + throw new InconsistentBrokerIdException("unable to match brokerId across logDirs") + } else if(brokerId >= 0 && metaBrokerIdSet.size == 1 && metaBrokerIdSet.last != brokerId) { + throw new InconsistentBrokerIdException("configured brokerId doesn't match stored brokerId in meta.properties") + } else if(metaBrokerIdSet.size == 0) { + if(brokerId < 0) { + brokerId = ZkUtils.getBrokerSequenceId(zkClient,config.MaxReservedBrokerId) + } else { + return brokerId + } + } else { + brokerId = metaBrokerIdSet.last + } + + if(!logDirsWithoutMetaProps.isEmpty) + storeBrokerId(brokerId, logDirsWithoutMetaProps) + + return brokerId + } + + private def readBrokerIdFromMetaProps(logDir: String): Option[Int] = { + try { + val metaProps = new VerifiableProperties(Utils.loadProps(logDir + File.separator + metaPropsFile)) + if (metaProps.containsKey("broker.id")) + return Some(metaProps.getIntInRange("broker.id", (0, Int.MaxValue))) + } catch { + case e: FileNotFoundException => + warn("unable to read meta.properties file under dir %s due to %s".format(logDir, e.getMessage)) + None + case e1: Exception => + error("unable to read meta.properties file under dir %s due to %s".format(logDir, e1.getMessage)) + throw e1 + } + None + } + + private def storeBrokerId(brokerId: Int, logDirs: Seq[String]) { + val metaProps = new Properties() + metaProps.setProperty("broker.id", brokerId.toString); + for(logDir <- logDirs) { + val f = Utils.createFile(logDir + File.separator + metaPropsFile) + val out = new FileOutputStream(f) + metaProps.store(out,"") + out.getFD().sync(); + out.close(); + } + } +} diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 897783c..6a56a77 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -36,9 +36,9 @@ object ReplicaManager { val HighWatermarkFilename = "replication-offset-checkpoint" } -class ReplicaManager(val config: KafkaConfig, - time: Time, - val zkClient: ZkClient, +class ReplicaManager(val config: KafkaConfig, + time: Time, + val zkClient: ZkClient, scheduler: Scheduler, val logManager: LogManager, val isShuttingDown: AtomicBoolean ) extends Logging with KafkaMetricsGroup { @@ -46,6 +46,8 @@ class ReplicaManager(val config: KafkaConfig, @volatile var controllerEpoch: Int = KafkaController.InitialControllerEpoch - 1 private val localBrokerId = config.brokerId private val allPartitions = new Pool[(String, Int), Partition] + private var leaderPartitions = new mutable.HashSet[Partition]() + private val leaderPartitionsLock = new Object private val replicaStateChangeLock = new Object val replicaFetcherManager = new ReplicaFetcherManager(config, this) private val highWatermarkCheckPointThreadStarted = new AtomicBoolean(false) @@ -58,7 +60,9 @@ class ReplicaManager(val config: KafkaConfig, "LeaderCount", new Gauge[Int] { def value = { - getLeaderPartitions().size + leaderPartitionsLock synchronized { + leaderPartitions.size + } } } ) @@ -78,7 +82,9 @@ class ReplicaManager(val config: KafkaConfig, val isrShrinkRate = newMeter("IsrShrinksPerSec", "shrinks", TimeUnit.SECONDS) def underReplicatedPartitionCount(): Int = { - getLeaderPartitions().count(_.isUnderReplicated) + leaderPartitionsLock synchronized { + leaderPartitions.count(_.isUnderReplicated) + } } def startHighWaterMarksCheckPointThread() = { @@ -111,6 +117,9 @@ class ReplicaManager(val config: KafkaConfig, val errorCode = ErrorMapping.NoError getPartition(topic, partitionId) match { case Some(partition) => + leaderPartitionsLock synchronized { + leaderPartitions -= partition + } if(deletePartition) { val removedPartition = allPartitions.remove((topic, partitionId)) if (removedPartition != null) @@ -322,6 +331,10 @@ class ReplicaManager(val config: KafkaConfig, partitionState.foreach{ case (partition, partitionStateInfo) => partition.makeLeader(controllerId, partitionStateInfo, correlationId, offsetManager)} + // Finally add these partitions to the list of partitions for which the leader is the current broker + leaderPartitionsLock synchronized { + leaderPartitions ++= partitionState.keySet + } } catch { case e: Throwable => partitionState.foreach { state => @@ -370,6 +383,9 @@ class ReplicaManager(val config: KafkaConfig, responseMap.put((partition.topic, partition.partitionId), ErrorMapping.NoError) try { + leaderPartitionsLock synchronized { + leaderPartitions --= partitionState.keySet + } var partitionsToMakeFollower: Set[Partition] = Set() @@ -448,7 +464,11 @@ class ReplicaManager(val config: KafkaConfig, private def maybeShrinkIsr(): Unit = { trace("Evaluating ISR list of partitions to see which replicas can be removed from the ISR") - allPartitions.values.foreach(partition => partition.maybeShrinkIsr(config.replicaLagTimeMaxMs, config.replicaLagMaxMessages)) + var curLeaderPartitions: List[Partition] = null + leaderPartitionsLock synchronized { + curLeaderPartitions = leaderPartitions.toList + } + curLeaderPartitions.foreach(partition => partition.maybeShrinkIsr(config.replicaLagTimeMaxMs, config.replicaLagMaxMessages)) } def recordFollowerPosition(topic: String, partitionId: Int, replicaId: Int, offset: Long) = { @@ -460,9 +480,6 @@ class ReplicaManager(val config: KafkaConfig, } } - private def getLeaderPartitions() : List[Partition] = { - allPartitions.values.filter(_.leaderReplicaIfLocal().isDefined).toList - } /** * Flushes the highwatermark value for all partitions to the highwatermark file */ diff --git a/core/src/main/scala/kafka/tools/DumpLogSegments.scala b/core/src/main/scala/kafka/tools/DumpLogSegments.scala index 8e9d47b..6daf87b 100644 --- a/core/src/main/scala/kafka/tools/DumpLogSegments.scala +++ b/core/src/main/scala/kafka/tools/DumpLogSegments.scala @@ -23,8 +23,7 @@ import kafka.log._ import kafka.utils._ import collection.mutable import joptsimple.OptionParser -import kafka.serializer.Decoder -import kafka.utils.VerifiableProperties + object DumpLogSegments { @@ -42,15 +41,7 @@ object DumpLogSegments { .ofType(classOf[java.lang.Integer]) .defaultsTo(5 * 1024 * 1024) val deepIterationOpt = parser.accepts("deep-iteration", "if set, uses deep instead of shallow iteration") - val valueDecoderOpt = parser.accepts("value-decoder-class", "if set, used to deserialize the messages. This class should implement kafka.serializer.Decoder trait. Custom jar should be available in kafka/libs directory.") - .withOptionalArg() - .ofType(classOf[java.lang.String]) - .defaultsTo("kafka.serializer.StringDecoder") - val keyDecoderOpt = parser.accepts("key-decoder-class", "if set, used to deserialize the keys. This class should implement kafka.serializer.Decoder trait. Custom jar should be available in kafka/libs directory.") - .withOptionalArg() - .ofType(classOf[java.lang.String]) - .defaultsTo("kafka.serializer.StringDecoder") - + if(args.length == 0) CommandLineUtils.printUsageAndDie(parser, "Parse a log file and dump its contents to the console, useful for debugging a seemingly corrupt log segment.") @@ -63,9 +54,6 @@ object DumpLogSegments { val files = options.valueOf(filesOpt).split(",") val maxMessageSize = options.valueOf(maxMessageSizeOpt).intValue() val isDeepIteration = if(options.has(deepIterationOpt)) true else false - - val valueDecoder: Decoder[_] = Utils.createObject[Decoder[_]](options.valueOf(valueDecoderOpt), new VerifiableProperties) - val keyDecoder: Decoder[_] = Utils.createObject[Decoder[_]](options.valueOf(keyDecoderOpt), new VerifiableProperties) val misMatchesForIndexFilesMap = new mutable.HashMap[String, List[(Long, Long)]] val nonConsecutivePairsForLogFilesMap = new mutable.HashMap[String, List[(Long, Long)]] @@ -74,7 +62,7 @@ object DumpLogSegments { val file = new File(arg) if(file.getName.endsWith(Log.LogFileSuffix)) { println("Dumping " + file) - dumpLog(file, print, nonConsecutivePairsForLogFilesMap, isDeepIteration, maxMessageSize , valueDecoder, keyDecoder) + dumpLog(file, print, nonConsecutivePairsForLogFilesMap, isDeepIteration, maxMessageSize) } else if(file.getName.endsWith(Log.IndexFileSuffix)) { println("Dumping " + file) dumpIndex(file, verifyOnly, misMatchesForIndexFilesMap, maxMessageSize) @@ -130,9 +118,7 @@ object DumpLogSegments { printContents: Boolean, nonConsecutivePairsForLogFilesMap: mutable.HashMap[String, List[(Long, Long)]], isDeepIteration: Boolean, - maxMessageSize: Int, - valueDecoder: Decoder[_], - keyDecoder: Decoder[_]) { + maxMessageSize: Int) { val startOffset = file.getName().split("\\.")(0).toLong println("Starting offset: " + startOffset) val messageSet = new FileMessageSet(file, false) @@ -161,8 +147,8 @@ object DumpLogSegments { print(" keysize: " + msg.keySize) if(printContents) { if(msg.hasKey) - print(" key: " + keyDecoder.fromBytes(Utils.readBytes(messageAndOffset.message.key))) - val payload = if(messageAndOffset.message.isNull) null else valueDecoder.fromBytes(Utils.readBytes(messageAndOffset.message.payload)) + print(" key: " + Utils.readString(messageAndOffset.message.key, "UTF-8")) + val payload = if(messageAndOffset.message.isNull) null else Utils.readString(messageAndOffset.message.payload, "UTF-8") print(" payload: " + payload) } println() @@ -200,5 +186,4 @@ object DumpLogSegments { } } } - } diff --git a/core/src/main/scala/kafka/utils/ZkUtils.scala b/core/src/main/scala/kafka/utils/ZkUtils.scala index dcdc1ce..30c7e3f 100644 --- a/core/src/main/scala/kafka/utils/ZkUtils.scala +++ b/core/src/main/scala/kafka/utils/ZkUtils.scala @@ -33,6 +33,7 @@ import kafka.controller.KafkaController import scala.Some import kafka.controller.LeaderIsrAndControllerEpoch import kafka.common.TopicAndPartition +import kafka.server.KafkaConfig import scala.collection object ZkUtils extends Logging { @@ -46,6 +47,8 @@ object ZkUtils extends Logging { val ReassignPartitionsPath = "/admin/reassign_partitions" val DeleteTopicsPath = "/admin/delete_topics" val PreferredReplicaLeaderElectionPath = "/admin/preferred_replica_election" + val BrokerSequenceIdPath = "/brokers/seqid" + val BrokerIdMaxValue = 1000 def getTopicPath(topic: String): String = { BrokerTopicsPath + "/" + topic @@ -87,7 +90,8 @@ object ZkUtils extends Logging { } def setupCommonPaths(zkClient: ZkClient) { - for(path <- Seq(ConsumersPath, BrokerIdsPath, BrokerTopicsPath, TopicConfigChangesPath, TopicConfigPath, DeleteTopicsPath)) + for(path <- Seq(ConsumersPath, BrokerIdsPath, BrokerTopicsPath, TopicConfigChangesPath, + TopicConfigPath, DeleteTopicsPath, BrokerSequenceIdPath)) makeSurePersistentPathExists(zkClient, path) } @@ -122,6 +126,14 @@ object ZkUtils extends Logging { } } + /** returns a sequence id generated by updating BrokerSequenceIdPath in Zk. + * users can provide brokerId in the config , inorder to avoid conflicts between zk generated + * seqId and config.brokerId we increment zk seqId by KafkaConfig.MaxReservedBrokerId. + */ + def getBrokerSequenceId(zkClient: ZkClient, MaxReservedBrokerId: Int): Int = { + getSequenceId(zkClient, BrokerSequenceIdPath) + MaxReservedBrokerId + } + /** * Gets the in-sync replicas (ISR) for a specific topic and partition */ @@ -691,6 +703,32 @@ object ZkUtils extends Logging { } } + /** + * This API produces a sequence number by creating / updating given path in zookeeper + * It uses the stat returned by the zookeeper and return the version. Every time + * client updates the path stat.version gets incremented + */ + def getSequenceId(client: ZkClient, path: String): Int = { + try { + val stat = client.writeDataReturnStat(path, "", -1) + return stat.getVersion + } catch { + case e: ZkNoNodeException => { + createParentPath(client, BrokerSequenceIdPath) + try { + client.createPersistent(BrokerSequenceIdPath, "") + return 0 + } catch { + case e: ZkNodeExistsException => + val stat = client.writeDataReturnStat(BrokerSequenceIdPath, "", -1) + return stat.getVersion + case e2: Throwable => throw e2 + } + } + case e2: Throwable => throw e2 + } + } + def getAllTopics(zkClient: ZkClient): Seq[String] = { val topics = ZkUtils.getChildrenParentMayNotExist(zkClient, BrokerTopicsPath) if(topics == null) diff --git a/core/src/test/scala/unit/kafka/server/ServerGenerateBrokerIdTest.scala b/core/src/test/scala/unit/kafka/server/ServerGenerateBrokerIdTest.scala new file mode 100644 index 0000000..80ecc9c --- /dev/null +++ b/core/src/test/scala/unit/kafka/server/ServerGenerateBrokerIdTest.scala @@ -0,0 +1,122 @@ +/** + * 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.server + +import kafka.zk.ZooKeeperTestHarness +import kafka.utils.{IntEncoder, TestUtils, Utils, VerifiableProperties} +import kafka.utils.TestUtils._ +import java.io.File +import org.junit.Test +import org.scalatest.junit.JUnit3Suite +import junit.framework.Assert._ + +class ServerGenerateBrokerIdTest extends JUnit3Suite with ZooKeeperTestHarness { + var props1 = TestUtils.createBrokerConfig(-1, TestUtils.choosePort) + var config1 = new KafkaConfig(props1) + var props2 = TestUtils.createBrokerConfig(0, TestUtils.choosePort) + var config2 = new KafkaConfig(props2) + + @Test + def testAutoGenerateBrokerId() { + var server1 = new KafkaServer(config1) + server1.startup() + // do a clean shutdown and check that offset checkpoint file exists + server1.shutdown() + for(logDir <- config1.logDirs) { + val metaProps = new VerifiableProperties(Utils.loadProps(logDir+"/meta.properties")) + assertTrue(metaProps.containsKey("broker.id")) + assertEquals(metaProps.getInt("broker.id"),1001) + } + // restart the server check to see if it uses the brokerId generated previously + server1 = new KafkaServer(config1) + server1.startup() + assertEquals(server1.config.brokerId, 1001) + server1.shutdown() + Utils.rm(server1.config.logDirs) + TestUtils.verifyNonDaemonThreadsStatus + } + + def testUserConfigAndGenratedBrokerId() { + // start the server with broker.id as part of config + val server1 = new KafkaServer(config1) + val server2 = new KafkaServer(config2) + val props3 = TestUtils.createBrokerConfig(-1, TestUtils.choosePort) + val config3 = new KafkaConfig(props3) + val server3 = new KafkaServer(config3) + server1.startup() + assertEquals(server1.config.brokerId,1001) + server2.startup() + assertEquals(server2.config.brokerId,0) + server3.startup() + assertEquals(server3.config.brokerId,1002) + server1.shutdown() + server2.shutdown() + server3.shutdown() + Utils.rm(server1.config.logDirs) + Utils.rm(server2.config.logDirs) + TestUtils.verifyNonDaemonThreadsStatus + } + + def testMultipleLogDirsMetaProps() { + // add multiple logDirs and check if the generate brokerId is stored in all of them + var logDirs = props1.getProperty("log.dir")+ "," + TestUtils.tempDir().getAbsolutePath + + "," + TestUtils.tempDir().getAbsolutePath + props1.setProperty("log.dir",logDirs) + config1 = new KafkaConfig(props1) + var server1 = new KafkaServer(config1) + server1.startup() + server1.shutdown() + for(logDir <- config1.logDirs) { + val metaProps = new VerifiableProperties(Utils.loadProps(logDir+"/meta.properties")) + assertTrue(metaProps.containsKey("broker.id")) + assertEquals(metaProps.getInt("broker.id"),1001) + } + + // addition to log.dirs after generation of a broker.id from zk should be copied over + var newLogDirs = props1.getProperty("log.dir") + "," + TestUtils.tempDir().getAbsolutePath + props1.setProperty("log.dir",newLogDirs) + config1 = new KafkaConfig(props1) + server1 = new KafkaServer(config1) + server1.startup() + server1.shutdown() + for(logDir <- config1.logDirs) { + val metaProps = new VerifiableProperties(Utils.loadProps(logDir+"/meta.properties")) + assertTrue(metaProps.containsKey("broker.id")) + assertEquals(metaProps.getInt("broker.id"),1001) + } + Utils.rm(server1.config.logDirs) + TestUtils.verifyNonDaemonThreadsStatus + } + + @Test + def testConsistentBrokerIdFromUserConfigAndMetaProps() { + // check if configured brokerId and stored brokerId are equal or throw InconsistentBrokerException + var server1 = new KafkaServer(config1) //auto generate broker Id + server1.startup() + server1.shutdown() + server1 = new KafkaServer(config2) // user specified broker id + try { + server1.startup() + } catch { + case e: kafka.common.InconsistentBrokerIdException => //success + } + server1.shutdown() + Utils.rm(server1.config.logDirs) + TestUtils.verifyNonDaemonThreadsStatus + } + +} diff --git a/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala b/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala index ab60e9b..e56ee15 100644 --- a/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala +++ b/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala @@ -63,7 +63,7 @@ class ServerShutdownTest extends JUnit3Suite with ZooKeeperTestHarness { assertTrue(OffsetCheckpointFile.length() > 0) } producer.close() - + /* now restart the server and check that the written data is still readable and everything still works */ server = new KafkaServer(config) server.startup() @@ -98,7 +98,7 @@ class ServerShutdownTest extends JUnit3Suite with ZooKeeperTestHarness { producer.close() server.shutdown() Utils.rm(server.config.logDirs) - verifyNonDaemonThreadsStatus + TestUtils.verifyNonDaemonThreadsStatus } @Test @@ -111,12 +111,7 @@ class ServerShutdownTest extends JUnit3Suite with ZooKeeperTestHarness { server.shutdown() server.awaitShutdown() Utils.rm(server.config.logDirs) - verifyNonDaemonThreadsStatus + TestUtils.verifyNonDaemonThreadsStatus } - def verifyNonDaemonThreadsStatus() { - assertEquals(0, Thread.getAllStackTraces.keySet().toArray - .map(_.asInstanceOf[Thread]) - .count(t => !t.isDaemon && t.isAlive && t.getClass.getCanonicalName.toLowerCase.startsWith("kafka"))) - } } diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index 4d01d25..e3c66bb 100644 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -90,7 +90,7 @@ object TestUtils extends Logging { Utils.rm(f) } }) - + f } @@ -150,7 +150,7 @@ object TestUtils extends Logging { def createBrokerConfig(nodeId: Int, port: Int = choosePort(), enableControlledShutdown: Boolean = true): Properties = { val props = new Properties - props.put("broker.id", nodeId.toString) + if (nodeId >= 0) props.put("broker.id", nodeId.toString) props.put("host.name", "localhost") props.put("port", port.toString) props.put("log.dir", TestUtils.tempDir().getAbsolutePath) @@ -689,6 +689,13 @@ object TestUtils extends Logging { def checkIfReassignPartitionPathExists(zkClient: ZkClient): Boolean = { ZkUtils.pathExists(zkClient, ZkUtils.ReassignPartitionsPath) } + + def verifyNonDaemonThreadsStatus() { + assertEquals(0, Thread.getAllStackTraces.keySet().toArray + .map(_.asInstanceOf[Thread]) + .count(t => !t.isDaemon && t.isAlive && t.getClass.getCanonicalName.toLowerCase.startsWith("kafka"))) + } + } object TestZKUtils {