From 2b43b11e7dc188010efdf3823b884f9b423f1edd Mon Sep 17 00:00:00 2001 From: Sriram Subramanian Date: Wed, 2 Jan 2013 17:06:33 -0800 Subject: [PATCH 01/10] Committer: Sriram Subramanian On branch 0.8 Your branch is behind 'origin/0.8' by 3 commits, and can be fast-forwarded. Changes to be committed: (use "git reset HEAD ..." to unstage) modified: core/src/main/scala/kafka/consumer/ConsumerConfig.scala modified: core/src/main/scala/kafka/log/LogManager.scala modified: core/src/main/scala/kafka/producer/ProducerConfig.scala modified: core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala modified: core/src/main/scala/kafka/server/KafkaConfig.scala modified: core/src/main/scala/kafka/server/KafkaServer.scala modified: core/src/main/scala/kafka/server/ReplicaFetcherThread.scala modified: core/src/main/scala/kafka/server/ReplicaManager.scala modified: core/src/test/scala/unit/kafka/log/LogManagerTest.scala modified: core/src/test/scala/unit/kafka/log/LogOffsetTest.scala modified: core/src/test/scala/unit/kafka/producer/ProducerTest.scala modified: core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala modified: core/src/test/scala/unit/kafka/utils/TestUtils.scala --- .../main/scala/kafka/consumer/ConsumerConfig.scala | 24 ++++++++++---------- core/src/main/scala/kafka/log/LogManager.scala | 20 ++++++++-------- .../main/scala/kafka/producer/ProducerConfig.scala | 2 +- .../kafka/producer/async/DefaultEventHandler.scala | 4 +- core/src/main/scala/kafka/server/KafkaConfig.scala | 3 +- core/src/main/scala/kafka/server/KafkaServer.scala | 4 +- .../scala/kafka/server/ReplicaFetcherThread.scala | 4 +- .../main/scala/kafka/server/ReplicaManager.scala | 2 +- .../test/scala/unit/kafka/log/LogManagerTest.scala | 14 +++++----- .../test/scala/unit/kafka/log/LogOffsetTest.scala | 2 +- .../scala/unit/kafka/producer/ProducerTest.scala | 2 +- .../scala/unit/kafka/server/LogRecoveryTest.scala | 2 +- .../test/scala/unit/kafka/utils/TestUtils.scala | 6 ++-- 13 files changed, 45 insertions(+), 44 deletions(-) diff --git a/core/src/main/scala/kafka/consumer/ConsumerConfig.scala b/core/src/main/scala/kafka/consumer/ConsumerConfig.scala index b379c9d..219df74 100644 --- a/core/src/main/scala/kafka/consumer/ConsumerConfig.scala +++ b/core/src/main/scala/kafka/consumer/ConsumerConfig.scala @@ -52,11 +52,11 @@ object ConsumerConfig extends Config { } def validateClientId(clientId: String) { - validateChars("clientid", clientId) + validateChars("client.id", clientId) } def validateGroupId(groupId: String) { - validateChars("groupid", groupId) + validateChars("group.id", groupId) } def validateAutoOffsetReset(autoOffsetReset: String) { @@ -77,32 +77,32 @@ class ConsumerConfig private (val props: VerifiableProperties) extends ZKConfig( } /** a string that uniquely identifies a set of consumers within the same consumer group */ - val groupId = props.getString("groupid") + val groupId = props.getString("group.id") /** consumer id: generated automatically if not set. * Set this explicitly for only testing purpose. */ - val consumerId: Option[String] = Option(props.getString("consumerid", null)) + val consumerId: Option[String] = Option(props.getString("consumer.id", null)) /** the socket timeout for network requests. The actual timeout set will be max.fetch.wait + socket.timeout.ms. */ val socketTimeoutMs = props.getInt("socket.timeout.ms", SocketTimeout) /** the socket receive buffer for network requests */ - val socketBufferSize = props.getInt("socket.buffersize", SocketBufferSize) + val socketBufferSize = props.getInt("socket.buffer.size", SocketBufferSize) /** the number of byes of messages to attempt to fetch */ val fetchSize = props.getInt("fetch.size", FetchSize) /** if true, periodically commit to zookeeper the offset of messages already fetched by the consumer */ - val autoCommit = props.getBoolean("autocommit.enable", AutoCommit) + val autoCommit = props.getBoolean("auto.commit.enable", AutoCommit) /** the frequency in ms that the consumer offsets are committed to zookeeper */ - val autoCommitIntervalMs = props.getInt("autocommit.interval.ms", AutoCommitInterval) + val autoCommitIntervalMs = props.getInt("auto.commit.interval.ms", AutoCommitInterval) /** max number of messages buffered for consumption */ - val maxQueuedChunks = props.getInt("queuedchunks.max", MaxQueuedChunks) + val maxQueuedChunks = props.getInt("max.queued.chunks", MaxQueuedChunks) /** max number of retries during rebalance */ - val maxRebalanceRetries = props.getInt("rebalance.retries.max", MaxRebalanceRetries) + val maxRebalanceRetries = props.getInt("max.rebalance.retries", MaxRebalanceRetries) /** the minimum amount of data the server should return for a fetch request. If insufficient data is available the request will block */ val minFetchBytes = props.getInt("min.fetch.bytes", MinFetchBytes) @@ -120,7 +120,7 @@ class ConsumerConfig private (val props: VerifiableProperties) extends ZKConfig( smallest : automatically reset the offset to the smallest offset largest : automatically reset the offset to the largest offset anything else: throw exception to the consumer */ - val autoOffsetReset = props.getString("autooffset.reset", AutoOffsetReset) + val autoOffsetReset = props.getString("auto.offset.reset", AutoOffsetReset) /** throw a timeout exception to the consumer if no message is available for consumption after the specified interval */ val consumerTimeoutMs = props.getInt("consumer.timeout.ms", ConsumerTimeoutMs) @@ -129,12 +129,12 @@ class ConsumerConfig private (val props: VerifiableProperties) extends ZKConfig( * Typically, it's only used for mirroring raw messages from one kafka cluster to another to save the * overhead of decompression. * */ - val enableShallowIterator = props.getBoolean("shallowiterator.enable", false) + val enableShallowIterator = props.getBoolean("enable.shallow.iterator", false) /** * Client id is specified by the kafka consumer client, used to distinguish different clients */ - val clientId = props.getString("clientid", groupId) + val clientId = props.getString("client.id", groupId) validate(this) } diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala index 5f0148c..c85f869 100644 --- a/core/src/main/scala/kafka/log/LogManager.scala +++ b/core/src/main/scala/kafka/log/LogManager.scala @@ -43,15 +43,15 @@ private[kafka] class LogManager(val config: KafkaConfig, val CleanShutdownFile = ".kafka_cleanshutdown" val LockFile = ".lock" val logDirs: Array[File] = config.logDirs.map(new File(_)).toArray - private val logFileSizeMap = config.logFileSizeMap - private val logFlushInterval = config.flushInterval - private val logFlushIntervals = config.flushIntervalMap + private val logFileSizeMap = config.logTopicFileSizeMap + private val logFlushInterval = config.logFlushInterval + private val logFlushIntervals = config.logTopicFlushIntervalMsMap private val logCreationLock = new Object - private val logRetentionSizeMap = config.logRetentionSizeMap - private val logRetentionMsMap = config.logRetentionHoursMap.map(e => (e._1, e._2 * 60 * 60 * 1000L)) // convert hours to ms - private val logRollMsMap = config.logRollHoursMap.map(e => (e._1, e._2 * 60 * 60 * 1000L)) + private val logRetentionSizeMap = config.logTopicRetentionSizeMap + private val logRetentionMsMap = config.logTopicRetentionHoursMap.map(e => (e._1, e._2 * 60 * 60 * 1000L)) // convert hours to ms + private val logRollMsMap = config.logTopicRollHoursMap.map(e => (e._1, e._2 * 60 * 60 * 1000L)) private val logRollDefaultIntervalMs = 1000L * 60 * 60 * config.logRollHours - private val logCleanupIntervalMs = 1000L * 60 * config.logCleanupIntervalMinutes + private val logCleanupIntervalMs = 1000L * 60 * config.logCleanupIntervalMins private val logCleanupDefaultAgeMs = 1000L * 60 * 60 * config.logRetentionHours this.logIdent = "[Log Manager on Broker " + config.brokerId + "] " @@ -139,10 +139,10 @@ private[kafka] class LogManager(val config: KafkaConfig, if(scheduler != null) { info("Starting log cleaner every " + logCleanupIntervalMs + " ms") scheduler.scheduleWithRate(cleanupLogs, "kafka-logcleaner-", 60 * 1000, logCleanupIntervalMs, false) - info("Starting log flusher every " + config.flushSchedulerThreadRate + + info("Starting log flusher every " + config.logFlushSchedulerIntervalMs + " ms with the following overrides " + logFlushIntervals) scheduler.scheduleWithRate(flushDirtyLogs, "kafka-logflusher-", - config.flushSchedulerThreadRate, config.flushSchedulerThreadRate, false) + config.logFlushSchedulerIntervalMs, config.logFlushSchedulerIntervalMs, false) } } @@ -310,7 +310,7 @@ private[kafka] class LogManager(val config: KafkaConfig, for (log <- allLogs) { try { val timeSinceLastFlush = System.currentTimeMillis - log.getLastFlushedTime - var logFlushInterval = config.defaultFlushIntervalMs + var logFlushInterval = config.logDefaultFlushIntervalMs if(logFlushIntervals.contains(log.topicName)) logFlushInterval = logFlushIntervals(log.topicName) debug(log.topicName + " flush interval " + logFlushInterval + diff --git a/core/src/main/scala/kafka/producer/ProducerConfig.scala b/core/src/main/scala/kafka/producer/ProducerConfig.scala index e559187..afab5d9 100644 --- a/core/src/main/scala/kafka/producer/ProducerConfig.scala +++ b/core/src/main/scala/kafka/producer/ProducerConfig.scala @@ -109,7 +109,7 @@ class ProducerConfig private (val props: VerifiableProperties) * ZK cache needs to be updated. * This parameter specifies the number of times the producer attempts to refresh this ZK cache. */ - val producerRetries = props.getInt("producer.num.retries", 3) + val producerRetryCount = props.getInt("producer.retry.count", 3) val producerRetryBackoffMs = props.getInt("producer.retry.backoff.ms", 100) diff --git a/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala b/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala index 4f04862..404c466 100644 --- a/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala +++ b/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala @@ -56,7 +56,7 @@ class DefaultEventHandler[K,V](config: ProducerConfig, producerTopicStats.getProducerAllTopicStats.byteRate.mark(dataSize) } var outstandingProduceRequests = serializedData - var remainingRetries = config.producerRetries + 1 + var remainingRetries = config.producerRetryCount + 1 while (remainingRetries > 0 && outstandingProduceRequests.size > 0) { outstandingProduceRequests = dispatchSerializedData(outstandingProduceRequests) if (outstandingProduceRequests.size > 0) { @@ -71,7 +71,7 @@ class DefaultEventHandler[K,V](config: ProducerConfig, if(outstandingProduceRequests.size > 0) { producerStats.failedSendRate.mark() error("Failed to send the following requests: " + outstandingProduceRequests) - throw new FailedToSendMessageException("Failed to send messages after " + config.producerRetries + " tries.", null) + throw new FailedToSendMessageException("Failed to send messages after " + config.producerRetryCount + " tries.", null) } } } diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index 962b65f..8dd3a4d 100644 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -20,6 +20,7 @@ package kafka.server import java.util.Properties import kafka.message.Message import kafka.consumer.ConsumerConfig +import java.net.InetAddress import kafka.utils.{VerifiableProperties, ZKConfig, Utils} /** @@ -57,7 +58,7 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro /* hostname of broker. If this is set, it will only bind to this address. If this is not set, * it will bind to all interfaces, and publish one to ZK */ - val hostName: String = props.getString("hostname", null) + val hostName: String = props.getString("host.name", null) /* the SO_SNDBUFF buffer of the socket sever sockets */ val socketSendBuffer: Int = props.getInt("socket.send.buffer", 100*1024) diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index ae35e4f..96dbe91 100644 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -65,8 +65,8 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime) extends Logg config.hostName, config.port, config.numNetworkThreads, - config.numQueuedRequests, - config.maxSocketRequestSize) + config.maxQueuedRequests, + config.maxSocketRequestBytes) socketServer.startup diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala index c1d3235..3b855b5 100644 --- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala +++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala @@ -34,8 +34,8 @@ class ReplicaFetcherThread(name:String, socketBufferSize = brokerConfig.replicaSocketBufferSize, fetchSize = brokerConfig.replicaFetchSize, fetcherBrokerId = brokerConfig.brokerId, - maxWait = brokerConfig.replicaMaxWaitTimeMs, - minBytes = brokerConfig.replicaMinBytes) { + maxWait = brokerConfig.replicaFetchMaxWaitTimeMs, + minBytes = brokerConfig.replicaFetchMinExpectedBytes) { // process fetched data def processPartitionData(topicAndPartition: TopicAndPartition, fetchOffset: Long, partitionData: FetchResponsePartitionData) { diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 42068ca..beef5b1 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -72,7 +72,7 @@ class ReplicaManager(val config: KafkaConfig, def startHighWaterMarksCheckPointThread() = { if(highWatermarkCheckPointThreadStarted.compareAndSet(false, true)) - kafkaScheduler.scheduleWithRate(checkpointHighWatermarks, "highwatermark-checkpoint-thread", 0, config.highWaterMarkCheckpointIntervalMs) + kafkaScheduler.scheduleWithRate(checkpointHighWatermarks, "highwatermark-checkpoint-thread", 0, config.replicaHighWatermarkCheckpointIntervalMs) } /** diff --git a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala index b06d812..c250312 100644 --- a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala @@ -41,7 +41,7 @@ class LogManagerTest extends JUnit3Suite { super.setUp() config = new KafkaConfig(TestUtils.createBrokerConfig(0, -1)) { override val logFileSize = 1024 - override val flushInterval = 10000 + override val logFlushInterval = 10000 override val logRetentionHours = maxLogAgeHours } scheduler.startup @@ -117,7 +117,7 @@ class LogManagerTest extends JUnit3Suite { override val logFileSize = (10 * (setSize - 1)) // each segment will be 10 messages override val logRetentionSize = (5 * 10 * setSize + 10).asInstanceOf[Long] override val logRetentionHours = retentionHours - override val flushInterval = 100 + override val logFlushInterval = 100 override val logRollHours = maxRollInterval } logManager = new LogManager(config, scheduler, time) @@ -159,10 +159,10 @@ class LogManagerTest extends JUnit3Suite { logManager.shutdown() config = new KafkaConfig(props) { override val logFileSize = 1024 *1024 *1024 - override val flushSchedulerThreadRate = 50 - override val flushInterval = Int.MaxValue + override val logFlushSchedulerIntervalMs = 50 + override val logFlushInterval = Int.MaxValue override val logRollHours = maxRollInterval - override val flushIntervalMap = Map("timebasedflush" -> 100) + override val logTopicFlushIntervalMsMap = Map("timebasedflush" -> 100) } logManager = new LogManager(config, scheduler, time) logManager.startup @@ -173,7 +173,7 @@ class LogManagerTest extends JUnit3Suite { } val ellapsed = System.currentTimeMillis - log.getLastFlushedTime assertTrue("The last flush time has to be within defaultflushInterval of current time (was %d)".format(ellapsed), - ellapsed < 2*config.flushSchedulerThreadRate) + ellapsed < 2*config.logFlushSchedulerIntervalMs) } @Test @@ -183,7 +183,7 @@ class LogManagerTest extends JUnit3Suite { val dirs = Seq(TestUtils.tempDir().getAbsolutePath, TestUtils.tempDir().getAbsolutePath, TestUtils.tempDir().getAbsolutePath) - props.put("log.directories", dirs.mkString(",")) + props.put("log.dirs", dirs.mkString(",")) logManager.shutdown() logManager = new LogManager(new KafkaConfig(props), scheduler, time) diff --git a/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala b/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala index c6ea3b6..dff6b24 100644 --- a/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala @@ -198,7 +198,7 @@ class LogOffsetTest extends JUnit3Suite with ZooKeeperTestHarness { private def createBrokerConfig(nodeId: Int, port: Int): Properties = { val props = new Properties - props.put("brokerid", nodeId.toString) + props.put("broker.id", nodeId.toString) props.put("port", port.toString) props.put("log.dir", getLogDir.getAbsolutePath) props.put("log.flush.interval", "1") diff --git a/core/src/test/scala/unit/kafka/producer/ProducerTest.scala b/core/src/test/scala/unit/kafka/producer/ProducerTest.scala index 48842eb..34ef470 100644 --- a/core/src/test/scala/unit/kafka/producer/ProducerTest.scala +++ b/core/src/test/scala/unit/kafka/producer/ProducerTest.scala @@ -300,7 +300,7 @@ class ProducerTest extends JUnit3Suite with ZooKeeperTestHarness with Logging{ // make sure we don't wait fewer than numRetries*timeoutMs milliseconds // we do this because the DefaultEventHandler retries a number of times - assertTrue((t2-t1) >= timeoutMs*config.producerRetries) + assertTrue((t2-t1) >= timeoutMs*config.producerRetryCount) } } diff --git a/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala b/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala index a3afa2d..dfdb0c3 100644 --- a/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala +++ b/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala @@ -30,7 +30,7 @@ class LogRecoveryTest extends JUnit3Suite with ZooKeeperTestHarness { val configs = TestUtils.createBrokerConfigs(2).map(new KafkaConfig(_) { override val replicaMaxLagTimeMs = 5000L override val replicaMaxLagBytes = 10L - override val replicaMinBytes = 20 + override val replicaFetchMinExpectedBytes = 20 }) val topic = "new-topic" val partitionId = 0 diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index a508895..ee4981b 100644 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -123,7 +123,7 @@ object TestUtils extends Logging { */ def createBrokerConfig(nodeId: Int, port: Int): Properties = { val props = new Properties - props.put("brokerid", nodeId.toString) + props.put("broker.id", nodeId.toString) props.put("hostname", "localhost") props.put("port", port.toString) props.put("log.dir", TestUtils.tempDir().getAbsolutePath) @@ -140,8 +140,8 @@ object TestUtils extends Logging { consumerTimeout: Long = -1): Properties = { val props = new Properties props.put("zk.connect", zkConnect) - props.put("groupid", groupId) - props.put("consumerid", consumerId) + props.put("group.id", groupId) + props.put("consumer.id", consumerId) props.put("consumer.timeout.ms", consumerTimeout.toString) props.put("zk.sessiontimeout.ms", "400") props.put("zk.synctime.ms", "200") -- 1.7.1 From 2c00843d3fdb83a83956103f70330255dcff11f8 Mon Sep 17 00:00:00 2001 From: Sriram Subramanian Date: Wed, 2 Jan 2013 18:06:28 -0800 Subject: [PATCH 02/10] Committer: Sriram Subramanian On branch 0.8 Your branch and 'origin/0.8' have diverged, and have 1 and 3 different commit(s) each, respectively. Changes to be committed: (use "git reset HEAD ..." to unstage) modified: config/consumer.properties modified: config/server.properties --- config/consumer.properties | 2 +- config/server.properties | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/config/consumer.properties b/config/consumer.properties index a067ac0..1c43bf9 100644 --- a/config/consumer.properties +++ b/config/consumer.properties @@ -23,7 +23,7 @@ zk.connect=127.0.0.1:2181 zk.connectiontimeout.ms=1000000 #consumer group id -groupid=test-consumer-group +group.id=test-consumer-group #consumer timeout #consumer.timeout.ms=5000 diff --git a/config/server.properties b/config/server.properties index f4521fb..4e14089 100644 --- a/config/server.properties +++ b/config/server.properties @@ -17,7 +17,7 @@ ############################# Server Basics ############################# # The id of the broker. This must be set to a unique integer for each broker. -brokerid=0 +broker.id=0 ############################# Socket Server Settings ############################# @@ -27,7 +27,7 @@ port=9092 # Hostname the broker will bind to and advertise to producers and consumers. # If not set, the server will bind to all interfaces and advertise the value returned from # from java.net.InetAddress.getCanonicalHostName(). -#hostname=localhost +#host.name=localhost # The number of threads handling network requests network.threads=2 @@ -78,7 +78,7 @@ log.default.flush.interval.ms=1000 #topic.flush.intervals.ms=topic1:1000, topic2:3000 # The interval (in ms) at which logs are checked to see if they need to be flushed to disk. -log.default.flush.scheduler.interval.ms=1000 +log.default.flush.interval.ms=1000 ############################# Log Retention Policy ############################# -- 1.7.1 From 2d20563b7251d2ce6ee2d3a6b8bd5519d0f1999e Mon Sep 17 00:00:00 2001 From: Sriram Subramanian Date: Thu, 3 Jan 2013 10:12:30 -0800 Subject: [PATCH 03/10] Committer: Sriram Subramanian On branch 0.8 Your branch is ahead of 'origin/0.8' by 2 commits. Changes to be committed: (use "git reset HEAD ..." to unstage) modified: core/src/main/scala/kafka/server/KafkaConfig.scala --- core/src/main/scala/kafka/server/KafkaConfig.scala | 49 ++++++++++---------- 1 files changed, 24 insertions(+), 25 deletions(-) diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index 8dd3a4d..c85f038 100644 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -37,28 +37,27 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro /*********** General Configuration ***********/ /* the broker id for this server */ - val brokerId: Int = props.getIntInRange("brokerid", (0, Int.MaxValue)) + val brokerId: Int = props.getIntInRange("broker.id", (0, Int.MaxValue)) /* the maximum size of message that the server can receive */ val maxMessageSize = props.getIntInRange("max.message.size", 1000000, (0, Int.MaxValue)) /* the number of network threads that the server uses for handling network requests */ - val numNetworkThreads = props.getIntInRange("network.threads", 3, (1, Int.MaxValue)) + val numNetworkThreads = props.getIntInRange("num.network.threads", 3, (1, Int.MaxValue)) /* the number of io threads that the server uses for carrying out network requests */ - val numIoThreads = props.getIntInRange("io.threads", 8, (1, Int.MaxValue)) + val numIoThreads = props.getIntInRange("num.io.threads", 8, (1, Int.MaxValue)) /* the number of queued requests allowed before blocking the network threads */ - val numQueuedRequests = props.getIntInRange("max.queued.requests", 500, (1, Int.MaxValue)) + val maxQueuedRequests = props.getIntInRange("max.queued.requests", 500, (1, Int.MaxValue)) /*********** Socket Server Configuration ***********/ /* the port to listen and accept connections on */ val port: Int = props.getInt("port", 6667) - /* hostname of broker. If this is set, it will only bind to this address. If this is not set, - * it will bind to all interfaces, and publish one to ZK */ - val hostName: String = props.getString("host.name", null) + /* hostname of broker. If not set, will pick up from the value returned from getLocalHost. If there are multiple interfaces getLocalHost may not be what you want. */ + val hostName: String = props.getString("host.name", InetAddress.getLocalHost.getHostAddress) /* the SO_SNDBUFF buffer of the socket sever sockets */ val socketSendBuffer: Int = props.getInt("socket.send.buffer", 100*1024) @@ -67,7 +66,7 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro val socketReceiveBuffer: Int = props.getInt("socket.receive.buffer", 100*1024) /* the maximum number of bytes in a socket request */ - val maxSocketRequestSize: Int = props.getIntInRange("max.socket.request.bytes", 100*1024*1024, (1, Int.MaxValue)) + val maxSocketRequestBytes: Int = props.getIntInRange("max.socket.request.bytes", 100*1024*1024, (1, Int.MaxValue)) /*********** Log Configuration ***********/ @@ -75,53 +74,53 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro val numPartitions = props.getIntInRange("num.partitions", 1, (1, Int.MaxValue)) /* the directories in which the log data is kept */ - val logDirs = Utils.parseCsvList(props.getString("log.directories", props.getString("log.dir", ""))) + val logDirs = Utils.parseCsvList(props.getString("log.dirs", props.getString("log.dir", ""))) require(logDirs.size > 0) /* the maximum size of a single log file */ val logFileSize = props.getIntInRange("log.file.size", 1*1024*1024*1024, (Message.MinHeaderSize, Int.MaxValue)) /* the maximum size of a single log file for some specific topic */ - val logFileSizeMap = props.getMap("topic.log.file.size", _.toInt > 0).mapValues(_.toInt) + val logTopicFileSizeMap = props.getMap("log.topic.file.size", _.toInt > 0).mapValues(_.toInt) /* the maximum time before a new log segment is rolled out */ val logRollHours = props.getIntInRange("log.roll.hours", 24*7, (1, Int.MaxValue)) /* the number of hours before rolling out a new log segment for some specific topic */ - val logRollHoursMap = props.getMap("topic.log.roll.hours", _.toInt > 0).mapValues(_.toInt) + val logTopicRollHoursMap = props.getMap("log.topic.roll.hours", _.toInt > 0).mapValues(_.toInt) /* the number of hours to keep a log file before deleting it */ val logRetentionHours = props.getIntInRange("log.retention.hours", 24*7, (1, Int.MaxValue)) /* the number of hours to keep a log file before deleting it for some specific topic*/ - val logRetentionHoursMap = props.getMap("topic.log.retention.hours", _.toInt > 0).mapValues(_.toInt) + val logTopicRetentionHoursMap = props.getMap("log.topic.retention.hours", _.toInt > 0).mapValues(_.toInt) /* the maximum size of the log before deleting it */ val logRetentionSize = props.getLong("log.retention.size", -1) /* the maximum size of the log for some specific topic before deleting it */ - val logRetentionSizeMap = props.getMap("topic.log.retention.size", _.toLong > 0).mapValues(_.toLong) + val logTopicRetentionSizeMap = props.getMap("log.topic.retention.size", _.toLong > 0).mapValues(_.toLong) /* the frequency in minutes that the log cleaner checks whether any log is eligible for deletion */ - val logCleanupIntervalMinutes = props.getIntInRange("log.cleanup.interval.mins", 10, (1, Int.MaxValue)) + val logCleanupIntervalMins = props.getIntInRange("log.cleanup.interval.mins", 10, (1, Int.MaxValue)) /* the maximum size in bytes of the offset index */ - val logIndexMaxSizeBytes = props.getIntInRange("log.index.max.size", 10*1024*1024, (4, Int.MaxValue)) + val logIndexMaxSizeBytes = props.getIntInRange("log.index.max.size.bytes", 10*1024*1024, (4, Int.MaxValue)) /* the interval with which we add an entry to the offset index */ val logIndexIntervalBytes = props.getIntInRange("log.index.interval.bytes", 4096, (0, Int.MaxValue)) /* the number of messages accumulated on a log partition before messages are flushed to disk */ - val flushInterval = props.getIntInRange("log.flush.interval", 500, (1, Int.MaxValue)) + val logFlushInterval = props.getIntInRange("log.flush.interval", 500, (1, Int.MaxValue)) /* the maximum time in ms that a message in selected topics is kept in memory before flushed to disk, e.g., topic1:3000,topic2: 6000 */ - val flushIntervalMap = props.getMap("topic.flush.intervals.ms", _.toInt > 0).mapValues(_.toInt) + val logTopicFlushIntervalMsMap = props.getMap("log.topic.flush.intervals.ms", _.toInt > 0).mapValues(_.toInt) /* the frequency in ms that the log flusher checks whether any log needs to be flushed to disk */ - val flushSchedulerThreadRate = props.getInt("log.default.flush.scheduler.interval.ms", 3000) + val logFlushSchedulerIntervalMs = props.getInt("log.flush.scheduler.interval.ms", 3000) /* the maximum time in ms that a message in any topic is kept in memory before flushed to disk */ - val defaultFlushIntervalMs = props.getInt("log.default.flush.interval.ms", flushSchedulerThreadRate) + val logDefaultFlushIntervalMs = props.getInt("log.default.flush.interval.ms", logFlushSchedulerIntervalMs) /* enable auto creation of topic on the server */ val autoCreateTopics = props.getBoolean("auto.create.topics", true) @@ -151,22 +150,22 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro val replicaFetchSize = props.getInt("replica.fetch.size", ConsumerConfig.FetchSize) /* max wait time for each fetcher request issued by follower replicas*/ - val replicaMaxWaitTimeMs = props.getInt("replica.fetch.wait.time.ms", 500) + val replicaFetchMaxWaitTimeMs = props.getInt("replica.fetch.max.wait.time.ms", 500) /* minimum bytes expected for each fetch response. If not enough bytes, wait up to replicaMaxWaitTimeMs */ - val replicaMinBytes = props.getInt("replica.fetch.min.bytes", 1) + val replicaFetchMinExpectedBytes = props.getInt("replica.fetch.min.expected.bytes", 1) /* number of fetcher threads used to replicate messages from a source broker. * Increasing this value can increase the degree of I/O parallelism in the follower broker. */ - val numReplicaFetchers = props.getInt("replica.fetchers", 1) + val numReplicaFetchers = props.getInt("num.replica.fetchers", 1) /* the frequency with which the highwater mark is saved out to disk */ - val highWaterMarkCheckpointIntervalMs = props.getLong("replica.highwatermark.checkpoint.ms", 5000L) + val replicaHighWatermarkCheckpointIntervalMs = props.getLong("replica.high.watermark.checkpoint.interval.ms", 5000L) /* the purge interval (in number of requests) of the fetch request purgatory */ - val fetchRequestPurgatoryPurgeInterval = props.getInt("fetch.purgatory.purge.interval", 10000) + val fetchRequestPurgatoryPurgeInterval = props.getInt("fetch.request.purgatory.purge.interval", 10000) /* the purge interval (in number of requests) of the producer request purgatory */ - val producerRequestPurgatoryPurgeInterval = props.getInt("producer.purgatory.purge.interval", 10000) + val producerRequestPurgatoryPurgeInterval = props.getInt("producer.request.purgatory.purge.interval", 10000) } -- 1.7.1 From f4b7ad8bc3d0acb39f1c15368a88ee3d1a43e43c Mon Sep 17 00:00:00 2001 From: Sriram Subramanian Date: Thu, 3 Jan 2013 11:03:15 -0800 Subject: [PATCH 04/10] Committer: Sriram Subramanian On branch 0.8 Your branch is ahead of 'origin/0.8' by 3 commits. Changes to be committed: (use "git reset HEAD ..." to unstage) modified: core/src/main/scala/kafka/server/KafkaConfig.scala --- core/src/main/scala/kafka/server/KafkaConfig.scala | 5 +++-- 1 files changed, 3 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index c85f038..7fc6ef5 100644 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -56,8 +56,9 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro /* the port to listen and accept connections on */ val port: Int = props.getInt("port", 6667) - /* hostname of broker. If not set, will pick up from the value returned from getLocalHost. If there are multiple interfaces getLocalHost may not be what you want. */ - val hostName: String = props.getString("host.name", InetAddress.getLocalHost.getHostAddress) + /* hostname of broker. If this is set, it will only bind to this address. If this is not set, + * it will bind to all interfaces, and publish one to ZK */ + val hostName: String = props.getString("host.name", null) /* the SO_SNDBUFF buffer of the socket sever sockets */ val socketSendBuffer: Int = props.getInt("socket.send.buffer", 100*1024) -- 1.7.1 From f55abce2987d9b91e3122912cc47b9cdb9641c33 Mon Sep 17 00:00:00 2001 From: Sriram Subramanian Date: Thu, 3 Jan 2013 11:54:00 -0800 Subject: [PATCH 05/10] Committer: Sriram Subramanian On branch 0.8 Your branch is ahead of 'origin/0.8' by 4 commits. Changes to be committed: (use "git reset HEAD ..." to unstage) modified: core/src/main/scala/kafka/server/KafkaConfig.scala modified: core/src/test/scala/unit/kafka/utils/TestUtils.scala --- core/src/main/scala/kafka/server/KafkaConfig.scala | 1 - .../test/scala/unit/kafka/utils/TestUtils.scala | 2 +- 2 files changed, 1 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index 7fc6ef5..8fccde5 100644 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -20,7 +20,6 @@ package kafka.server import java.util.Properties import kafka.message.Message import kafka.consumer.ConsumerConfig -import java.net.InetAddress import kafka.utils.{VerifiableProperties, ZKConfig, Utils} /** diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index ee4981b..dec245a 100644 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -124,7 +124,7 @@ object TestUtils extends Logging { def createBrokerConfig(nodeId: Int, port: Int): Properties = { val props = new Properties props.put("broker.id", nodeId.toString) - props.put("hostname", "localhost") + props.put("host.name", "localhost") props.put("port", port.toString) props.put("log.dir", TestUtils.tempDir().getAbsolutePath) props.put("log.flush.interval", "1") -- 1.7.1 From f404c4454f20593e5692b9fc799831e578de3c01 Mon Sep 17 00:00:00 2001 From: Sriram Subramanian Date: Thu, 3 Jan 2013 13:52:45 -0800 Subject: [PATCH 06/10] KAFKA-648 Use uniform convention for naming properties keys On branch 0.8 Your branch is ahead of 'origin/0.8' by 5 commits. Changes to be committed: (use "git reset HEAD ..." to unstage) modified: config/server.properties modified: core/src/main/scala/kafka/log/LogManager.scala modified: core/src/main/scala/kafka/server/KafkaConfig.scala modified: core/src/test/scala/unit/kafka/log/LogManagerTest.scala modified: core/src/test/scala/unit/kafka/log/LogOffsetTest.scala --- config/server.properties | 2 +- core/src/main/scala/kafka/log/LogManager.scala | 14 +++++++------- core/src/main/scala/kafka/server/KafkaConfig.scala | 12 ++++++------ .../test/scala/unit/kafka/log/LogManagerTest.scala | 8 ++++---- .../test/scala/unit/kafka/log/LogOffsetTest.scala | 2 +- 5 files changed, 19 insertions(+), 19 deletions(-) diff --git a/config/server.properties b/config/server.properties index 4e14089..241e46e 100644 --- a/config/server.properties +++ b/config/server.properties @@ -95,7 +95,7 @@ log.retention.hours=168 #log.retention.size=1073741824 # The maximum size of a log segment file. When this size is reached a new log segment will be created. -log.file.size=536870912 +log.segment.size=536870912 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala index c85f869..e4c948f 100644 --- a/core/src/main/scala/kafka/log/LogManager.scala +++ b/core/src/main/scala/kafka/log/LogManager.scala @@ -43,13 +43,13 @@ private[kafka] class LogManager(val config: KafkaConfig, val CleanShutdownFile = ".kafka_cleanshutdown" val LockFile = ".lock" val logDirs: Array[File] = config.logDirs.map(new File(_)).toArray - private val logFileSizeMap = config.logTopicFileSizeMap + private val logFileSizeMap = config.logSegmentSizePerTopicMap private val logFlushInterval = config.logFlushInterval - private val logFlushIntervals = config.logTopicFlushIntervalMsMap + private val logFlushIntervals = config.logFlushIntervalMsPerTopicMap private val logCreationLock = new Object - private val logRetentionSizeMap = config.logTopicRetentionSizeMap - private val logRetentionMsMap = config.logTopicRetentionHoursMap.map(e => (e._1, e._2 * 60 * 60 * 1000L)) // convert hours to ms - private val logRollMsMap = config.logTopicRollHoursMap.map(e => (e._1, e._2 * 60 * 60 * 1000L)) + private val logRetentionSizeMap = config.logRetentionSizePerTopicMap + private val logRetentionMsMap = config.logRetentionHoursPerTopicMap.map(e => (e._1, e._2 * 60 * 60 * 1000L)) // convert hours to ms + private val logRollMsMap = config.logRollHoursPerTopicMap.map(e => (e._1, e._2 * 60 * 60 * 1000L)) private val logRollDefaultIntervalMs = 1000L * 60 * 60 * config.logRollHours private val logCleanupIntervalMs = 1000L * 60 * config.logCleanupIntervalMins private val logCleanupDefaultAgeMs = 1000L * 60 * 60 * config.logRetentionHours @@ -111,7 +111,7 @@ private[kafka] class LogManager(val config: KafkaConfig, info("Loading log '" + dir.getName + "'") val topicPartition = parseTopicPartitionName(dir.getName) val rollIntervalMs = logRollMsMap.get(topicPartition.topic).getOrElse(this.logRollDefaultIntervalMs) - val maxLogFileSize = logFileSizeMap.get(topicPartition.topic).getOrElse(config.logFileSize) + val maxLogFileSize = logFileSizeMap.get(topicPartition.topic).getOrElse(config.logSegmentSize) val log = new Log(dir, maxLogFileSize, config.maxMessageSize, @@ -186,7 +186,7 @@ private[kafka] class LogManager(val config: KafkaConfig, val dir = new File(dataDir, topicAndPartition.topic + "-" + topicAndPartition.partition) dir.mkdirs() val rollIntervalMs = logRollMsMap.get(topicAndPartition.topic).getOrElse(this.logRollDefaultIntervalMs) - val maxLogFileSize = logFileSizeMap.get(topicAndPartition.topic).getOrElse(config.logFileSize) + val maxLogFileSize = logFileSizeMap.get(topicAndPartition.topic).getOrElse(config.logSegmentSize) log = new Log(dir, maxLogFileSize, config.maxMessageSize, diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index 8fccde5..12592eb 100644 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -78,28 +78,28 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro require(logDirs.size > 0) /* the maximum size of a single log file */ - val logFileSize = props.getIntInRange("log.file.size", 1*1024*1024*1024, (Message.MinHeaderSize, Int.MaxValue)) + val logSegmentSize = props.getIntInRange("log.segment.size", 1*1024*1024*1024, (Message.MinHeaderSize, Int.MaxValue)) /* the maximum size of a single log file for some specific topic */ - val logTopicFileSizeMap = props.getMap("log.topic.file.size", _.toInt > 0).mapValues(_.toInt) + val logSegmentSizePerTopicMap = props.getMap("log.segment.size.per.topic", _.toInt > 0).mapValues(_.toInt) /* the maximum time before a new log segment is rolled out */ val logRollHours = props.getIntInRange("log.roll.hours", 24*7, (1, Int.MaxValue)) /* the number of hours before rolling out a new log segment for some specific topic */ - val logTopicRollHoursMap = props.getMap("log.topic.roll.hours", _.toInt > 0).mapValues(_.toInt) + val logRollHoursPerTopicMap = props.getMap("log.roll.hours.per.topic", _.toInt > 0).mapValues(_.toInt) /* the number of hours to keep a log file before deleting it */ val logRetentionHours = props.getIntInRange("log.retention.hours", 24*7, (1, Int.MaxValue)) /* the number of hours to keep a log file before deleting it for some specific topic*/ - val logTopicRetentionHoursMap = props.getMap("log.topic.retention.hours", _.toInt > 0).mapValues(_.toInt) + val logRetentionHoursPerTopicMap = props.getMap("log.retention.hours.per.topic", _.toInt > 0).mapValues(_.toInt) /* the maximum size of the log before deleting it */ val logRetentionSize = props.getLong("log.retention.size", -1) /* the maximum size of the log for some specific topic before deleting it */ - val logTopicRetentionSizeMap = props.getMap("log.topic.retention.size", _.toLong > 0).mapValues(_.toLong) + val logRetentionSizePerTopicMap = props.getMap("log.retention.size.per.topic", _.toLong > 0).mapValues(_.toLong) /* the frequency in minutes that the log cleaner checks whether any log is eligible for deletion */ val logCleanupIntervalMins = props.getIntInRange("log.cleanup.interval.mins", 10, (1, Int.MaxValue)) @@ -114,7 +114,7 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro val logFlushInterval = props.getIntInRange("log.flush.interval", 500, (1, Int.MaxValue)) /* the maximum time in ms that a message in selected topics is kept in memory before flushed to disk, e.g., topic1:3000,topic2: 6000 */ - val logTopicFlushIntervalMsMap = props.getMap("log.topic.flush.intervals.ms", _.toInt > 0).mapValues(_.toInt) + val logFlushIntervalMsPerTopicMap = props.getMap("log.flush.intervals.ms.per.topic", _.toInt > 0).mapValues(_.toInt) /* the frequency in ms that the log flusher checks whether any log needs to be flushed to disk */ val logFlushSchedulerIntervalMs = props.getInt("log.flush.scheduler.interval.ms", 3000) diff --git a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala index c250312..ac5e38b 100644 --- a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala @@ -40,7 +40,7 @@ class LogManagerTest extends JUnit3Suite { override def setUp() { super.setUp() config = new KafkaConfig(TestUtils.createBrokerConfig(0, -1)) { - override val logFileSize = 1024 + override val logSegmentSize = 1024 override val logFlushInterval = 10000 override val logRetentionHours = maxLogAgeHours } @@ -114,7 +114,7 @@ class LogManagerTest extends JUnit3Suite { val props = TestUtils.createBrokerConfig(0, -1) logManager.shutdown() config = new KafkaConfig(props) { - override val logFileSize = (10 * (setSize - 1)) // each segment will be 10 messages + override val logSegmentSize = (10 * (setSize - 1)) // each segment will be 10 messages override val logRetentionSize = (5 * 10 * setSize + 10).asInstanceOf[Long] override val logRetentionHours = retentionHours override val logFlushInterval = 100 @@ -158,11 +158,11 @@ class LogManagerTest extends JUnit3Suite { val props = TestUtils.createBrokerConfig(0, -1) logManager.shutdown() config = new KafkaConfig(props) { - override val logFileSize = 1024 *1024 *1024 + override val logSegmentSize = 1024 *1024 *1024 override val logFlushSchedulerIntervalMs = 50 override val logFlushInterval = Int.MaxValue override val logRollHours = maxRollInterval - override val logTopicFlushIntervalMsMap = Map("timebasedflush" -> 100) + override val logFlushIntervalMsPerTopicMap = Map("timebasedflush" -> 100) } logManager = new LogManager(config, scheduler, time) logManager.startup diff --git a/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala b/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala index dff6b24..e33623d 100644 --- a/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala @@ -206,7 +206,7 @@ class LogOffsetTest extends JUnit3Suite with ZooKeeperTestHarness { props.put("num.partitions", "20") props.put("log.retention.hours", "10") props.put("log.cleanup.interval.mins", "5") - props.put("log.file.size", logSize.toString) + props.put("log.segment.size", logSize.toString) props.put("zk.connect", zkConnect.toString) props } -- 1.7.1 From 60dad2db42fd90fce98b34f0904137d241a2ebd3 Mon Sep 17 00:00:00 2001 From: Sriram Subramanian Date: Mon, 7 Jan 2013 15:08:09 -0800 Subject: [PATCH 07/10] Kafka 648 - Use uniform convention for naming properties keys modified: config/producer.properties modified: core/src/main/scala/kafka/consumer/ConsumerConfig.scala modified: core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala modified: core/src/main/scala/kafka/producer/KafkaLog4jAppender.scala modified: core/src/main/scala/kafka/producer/Producer.scala modified: core/src/main/scala/kafka/producer/ProducerConfig.scala modified: core/src/main/scala/kafka/producer/SyncProducer.scala modified: core/src/main/scala/kafka/producer/SyncProducerConfig.scala modified: core/src/main/scala/kafka/producer/async/AsyncProducerConfig.scala modified: core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala modified: core/src/main/scala/kafka/tools/ReplayLogProducer.scala modified: perf/src/main/scala/kafka/perf/ProducerPerformance.scala modified: system_test/broker_failure/config/mirror_producer.properties modified: system_test/broker_failure/config/mirror_producer1.properties modified: system_test/broker_failure/config/mirror_producer2.properties modified: system_test/broker_failure/config/mirror_producer3.properties modified: system_test/migration_tool_testsuite/config/migration_producer.properties modified: system_test/mirror_maker/config/mirror_producer.properties modified: system_test/mirror_maker_testsuite/config/mirror_producer.properties --- config/producer.properties | 2 +- .../main/scala/kafka/consumer/ConsumerConfig.scala | 4 ++-- .../consumer/ZookeeperConsumerConnector.scala | 10 +++++----- .../scala/kafka/producer/KafkaLog4jAppender.scala | 2 +- core/src/main/scala/kafka/producer/Producer.scala | 6 +++--- .../main/scala/kafka/producer/ProducerConfig.scala | 2 +- .../main/scala/kafka/producer/SyncProducer.scala | 2 +- .../scala/kafka/producer/SyncProducerConfig.scala | 6 +++--- .../kafka/producer/async/AsyncProducerConfig.scala | 2 +- .../kafka/producer/async/DefaultEventHandler.scala | 4 ++-- .../main/scala/kafka/tools/ReplayLogProducer.scala | 2 +- .../scala/kafka/perf/ProducerPerformance.scala | 2 +- .../config/mirror_producer.properties | 2 +- .../config/mirror_producer1.properties | 2 +- .../config/mirror_producer2.properties | 2 +- .../config/mirror_producer3.properties | 2 +- .../config/migration_producer.properties | 2 +- .../mirror_maker/config/mirror_producer.properties | 2 +- .../config/mirror_producer.properties | 2 +- 19 files changed, 29 insertions(+), 29 deletions(-) diff --git a/config/producer.properties b/config/producer.properties index eb36691..650f80c 100644 --- a/config/producer.properties +++ b/config/producer.properties @@ -51,7 +51,7 @@ serializer.class=kafka.serializer.StringEncoder # 0: events will be enqueued immediately or dropped if the queue is full # -ve: enqueue will block indefinitely if the queue is full # +ve: enqueue will block up to this many milliseconds if the queue is full -#queue.enqueueTimeout.ms= +#queue.enqueue.timeout.ms= # the number of messages batched at the producer #batch.size= diff --git a/core/src/main/scala/kafka/consumer/ConsumerConfig.scala b/core/src/main/scala/kafka/consumer/ConsumerConfig.scala index 219df74..26c3dd7 100644 --- a/core/src/main/scala/kafka/consumer/ConsumerConfig.scala +++ b/core/src/main/scala/kafka/consumer/ConsumerConfig.scala @@ -93,7 +93,7 @@ class ConsumerConfig private (val props: VerifiableProperties) extends ZKConfig( val fetchSize = props.getInt("fetch.size", FetchSize) /** if true, periodically commit to zookeeper the offset of messages already fetched by the consumer */ - val autoCommit = props.getBoolean("auto.commit.enable", AutoCommit) + val autoCommitEnable = props.getBoolean("auto.commit.enable", AutoCommit) /** the frequency in ms that the consumer offsets are committed to zookeeper */ val autoCommitIntervalMs = props.getInt("auto.commit.interval.ms", AutoCommitInterval) @@ -129,7 +129,7 @@ class ConsumerConfig private (val props: VerifiableProperties) extends ZKConfig( * Typically, it's only used for mirroring raw messages from one kafka cluster to another to save the * overhead of decompression. * */ - val enableShallowIterator = props.getBoolean("enable.shallow.iterator", false) + val shallowIteratorEnable = props.getBoolean("shallow.iterator.enable", false) /** * Client id is specified by the kafka consumer client, used to distinguish different clients diff --git a/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala b/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala index aee9293..ed41308 100644 --- a/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala +++ b/core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala @@ -112,7 +112,7 @@ private[kafka] class ZookeeperConsumerConnector(val config: ConsumerConfig, connectZk() createFetcher() - if (config.autoCommit) { + if (config.autoCommitEnable) { scheduler.startup info("starting auto committer every " + config.autoCommitIntervalMs + " ms") scheduler.scheduleWithRate(autoCommit, "Kafka-consumer-autocommit-", config.autoCommitIntervalMs, @@ -160,14 +160,14 @@ private[kafka] class ZookeeperConsumerConnector(val config: ConsumerConfig, if (wildcardTopicWatcher != null) wildcardTopicWatcher.shutdown() try { - if (config.autoCommit) + if (config.autoCommitEnable) scheduler.shutdownNow() fetcher match { case Some(f) => f.shutdown case None => } sendShutdownToAllQueues() - if (config.autoCommit) + if (config.autoCommitEnable) commitOffsets() if (zkClient != null) { zkClient.close() @@ -196,7 +196,7 @@ private[kafka] class ZookeeperConsumerConnector(val config: ConsumerConfig, threadIdSet.map(_ => { val queue = new LinkedBlockingQueue[FetchedDataChunk](config.maxQueuedChunks) val stream = new KafkaStream[K,V]( - queue, config.consumerTimeoutMs, keyDecoder, valueDecoder, config.enableShallowIterator, config.clientId) + queue, config.consumerTimeoutMs, keyDecoder, valueDecoder, config.shallowIteratorEnable, config.clientId) (queue, stream) }) ).flatten.toList @@ -714,7 +714,7 @@ private[kafka] class ZookeeperConsumerConnector(val config: ConsumerConfig, config.consumerTimeoutMs, keyDecoder, valueDecoder, - config.enableShallowIterator, + config.shallowIteratorEnable, config.clientId) (queue, stream) }).toList diff --git a/core/src/main/scala/kafka/producer/KafkaLog4jAppender.scala b/core/src/main/scala/kafka/producer/KafkaLog4jAppender.scala index a7c101a..c7f9d42 100644 --- a/core/src/main/scala/kafka/producer/KafkaLog4jAppender.scala +++ b/core/src/main/scala/kafka/producer/KafkaLog4jAppender.scala @@ -73,7 +73,7 @@ class KafkaLog4jAppender extends AppenderSkeleton with Logging { //These have default values in ProducerConfig and AsyncProducerConfig. We don't care if they're not specified if(producerType != null) props.put("producer.type", producerType) if(compressionCodec != null) props.put("compression.codec", compressionCodec) - if(enqueueTimeout != null) props.put("queue.enqueueTimeout.ms", enqueueTimeout) + if(enqueueTimeout != null) props.put("queue.enqueue.timeout.ms", enqueueTimeout) if(queueSize != null) props.put("queue.size", queueSize) val config : ProducerConfig = new ProducerConfig(props) producer = new Producer[String, String](config) diff --git a/core/src/main/scala/kafka/producer/Producer.scala b/core/src/main/scala/kafka/producer/Producer.scala index a183525..513468a 100644 --- a/core/src/main/scala/kafka/producer/Producer.scala +++ b/core/src/main/scala/kafka/producer/Producer.scala @@ -87,17 +87,17 @@ class Producer[K,V](config: ProducerConfig, private def asyncSend(messages: Seq[KeyedMessage[K,V]]) { for (message <- messages) { - val added = config.enqueueTimeoutMs match { + val added = config.queueEnqueueTimeoutMs match { case 0 => queue.offer(message) case _ => try { - config.enqueueTimeoutMs < 0 match { + config.queueEnqueueTimeoutMs < 0 match { case true => queue.put(message) true case _ => - queue.offer(message, config.enqueueTimeoutMs, TimeUnit.MILLISECONDS) + queue.offer(message, config.queueEnqueueTimeoutMs, TimeUnit.MILLISECONDS) } } catch { diff --git a/core/src/main/scala/kafka/producer/ProducerConfig.scala b/core/src/main/scala/kafka/producer/ProducerConfig.scala index afab5d9..6992813 100644 --- a/core/src/main/scala/kafka/producer/ProducerConfig.scala +++ b/core/src/main/scala/kafka/producer/ProducerConfig.scala @@ -31,7 +31,7 @@ object ProducerConfig extends Config { } def validateClientId(clientId: String) { - validateChars("clientid", clientId) + validateChars("client.id", clientId) } def validateBatchSize(batchSize: Int, queueSize: Int) { diff --git a/core/src/main/scala/kafka/producer/SyncProducer.scala b/core/src/main/scala/kafka/producer/SyncProducer.scala index 0ef320b..acb8687 100644 --- a/core/src/main/scala/kafka/producer/SyncProducer.scala +++ b/core/src/main/scala/kafka/producer/SyncProducer.scala @@ -36,7 +36,7 @@ class SyncProducer(val config: SyncProducerConfig) extends Logging { private val lock = new Object() @volatile private var shutdown: Boolean = false private val blockingChannel = new BlockingChannel(config.host, config.port, BlockingChannel.UseDefaultBufferSize, - config.bufferSize, config.requestTimeoutMs) + config.bufferSize, config.producerRequestTimeoutMs) val brokerInfo = "host_%s-port_%s".format(config.host, config.port) val producerRequestStats = ProducerRequestStatsRegistry.getProducerRequestStats(config.clientId) diff --git a/core/src/main/scala/kafka/producer/SyncProducerConfig.scala b/core/src/main/scala/kafka/producer/SyncProducerConfig.scala index 5ebd29a..7fdd38b 100644 --- a/core/src/main/scala/kafka/producer/SyncProducerConfig.scala +++ b/core/src/main/scala/kafka/producer/SyncProducerConfig.scala @@ -41,19 +41,19 @@ trait SyncProducerConfigShared { val maxMessageSize = props.getInt("max.message.size", 1000000) /* the client application sending the producer requests */ - val clientId = props.getString("clientid", SyncProducerConfig.DefaultClientId) + val clientId = props.getString("client.id", SyncProducerConfig.DefaultClientId) /* * The required acks of the producer requests - negative value means ack * after the replicas in ISR have caught up to the leader's offset * corresponding to this produce request. */ - val requiredAcks = props.getShort("producer.request.required.acks", SyncProducerConfig.DefaultRequiredAcks) + val producerRequestRequiredAcks = props.getShort("producer.request.required.acks", SyncProducerConfig.DefaultRequiredAcks) /* * The ack timeout of the producer requests. Value must be non-negative and non-zero */ - val requestTimeoutMs = props.getIntInRange("producer.request.timeout.ms", SyncProducerConfig.DefaultAckTimeoutMs, + val producerRequestTimeoutMs = props.getIntInRange("producer.request.timeout.ms", SyncProducerConfig.DefaultAckTimeoutMs, (1, Integer.MAX_VALUE)) } diff --git a/core/src/main/scala/kafka/producer/async/AsyncProducerConfig.scala b/core/src/main/scala/kafka/producer/async/AsyncProducerConfig.scala index 07935d7..4fff6c8 100644 --- a/core/src/main/scala/kafka/producer/async/AsyncProducerConfig.scala +++ b/core/src/main/scala/kafka/producer/async/AsyncProducerConfig.scala @@ -33,7 +33,7 @@ trait AsyncProducerConfig { * -ve: enqueue will block indefinitely if the queue is full * +ve: enqueue will block up to this many milliseconds if the queue is full */ - val enqueueTimeoutMs = props.getInt("queue.enqueueTimeout.ms", 0) + val queueEnqueueTimeoutMs = props.getInt("queue.enqueue.timeout.ms", 0) /** the number of messages batched at the producer */ val batchSize = props.getInt("batch.size", 200) diff --git a/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala b/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala index 404c466..58cb1d0 100644 --- a/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala +++ b/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala @@ -214,8 +214,8 @@ class DefaultEventHandler[K,V](config: ProducerConfig, warn("Failed to send to broker %d with data %s".format(brokerId, messagesPerTopic)) messagesPerTopic.keys.toSeq } else if(messagesPerTopic.size > 0) { - val producerRequest = new ProducerRequest(correlationCounter.getAndIncrement(), config.clientId, config.requiredAcks, - config.requestTimeoutMs, messagesPerTopic) + val producerRequest = new ProducerRequest(correlationCounter.getAndIncrement(), config.clientId, config.producerRequestRequiredAcks, + config.producerRequestTimeoutMs, messagesPerTopic) try { val syncProducer = producerPool.getProducer(brokerId) val response = syncProducer.send(producerRequest) diff --git a/core/src/main/scala/kafka/tools/ReplayLogProducer.scala b/core/src/main/scala/kafka/tools/ReplayLogProducer.scala index db14c82..4cf0a73 100644 --- a/core/src/main/scala/kafka/tools/ReplayLogProducer.scala +++ b/core/src/main/scala/kafka/tools/ReplayLogProducer.scala @@ -144,7 +144,7 @@ object ReplayLogProducer extends Logging { props.put("buffer.size", (64*1024).toString) props.put("compression.codec", config.compressionCodec.codec.toString) props.put("batch.size", config.batchSize.toString) - props.put("queue.enqueueTimeout.ms", "-1") + props.put("queue.enqueue.timeout.ms", "-1") if(config.isAsync) props.put("producer.type", "async") diff --git a/perf/src/main/scala/kafka/perf/ProducerPerformance.scala b/perf/src/main/scala/kafka/perf/ProducerPerformance.scala index a9a5f07..1942a38 100644 --- a/perf/src/main/scala/kafka/perf/ProducerPerformance.scala +++ b/perf/src/main/scala/kafka/perf/ProducerPerformance.scala @@ -195,7 +195,7 @@ object ProducerPerformance extends Logging { if(!config.isSync) { props.put("producer.type","async") props.put("batch.size", config.batchSize.toString) - props.put("queue.enqueueTimeout.ms", "-1") + props.put("queue.enqueue.timeout.ms", "-1") } props.put("producer.request.required.acks", config.producerRequestRequiredAcks.toString) props.put("producer.request.timeout.ms", config.producerRequestTimeoutMs.toString) diff --git a/system_test/broker_failure/config/mirror_producer.properties b/system_test/broker_failure/config/mirror_producer.properties index 9ea68d0..af5d4df 100644 --- a/system_test/broker_failure/config/mirror_producer.properties +++ b/system_test/broker_failure/config/mirror_producer.properties @@ -23,5 +23,5 @@ zk.connectiontimeout.ms=1000000 producer.type=async # to avoid dropping events if the queue is full, wait indefinitely -queue.enqueueTimeout.ms=-1 +queue.enqueue.timeout.ms=-1 diff --git a/system_test/broker_failure/config/mirror_producer1.properties b/system_test/broker_failure/config/mirror_producer1.properties index 7f37db3..ac24e7f 100644 --- a/system_test/broker_failure/config/mirror_producer1.properties +++ b/system_test/broker_failure/config/mirror_producer1.properties @@ -24,5 +24,5 @@ zk.connectiontimeout.ms=1000000 producer.type=async # to avoid dropping events if the queue is full, wait indefinitely -queue.enqueueTimeout.ms=-1 +queue.enqueue.timeout.ms=-1 diff --git a/system_test/broker_failure/config/mirror_producer2.properties b/system_test/broker_failure/config/mirror_producer2.properties index 047f840..3c0df5c 100644 --- a/system_test/broker_failure/config/mirror_producer2.properties +++ b/system_test/broker_failure/config/mirror_producer2.properties @@ -24,5 +24,5 @@ zk.connectiontimeout.ms=1000000 producer.type=async # to avoid dropping events if the queue is full, wait indefinitely -queue.enqueueTimeout.ms=-1 +queue.enqueue.timeout.ms=-1 diff --git a/system_test/broker_failure/config/mirror_producer3.properties b/system_test/broker_failure/config/mirror_producer3.properties index 5e8b7dc..5f525d3 100644 --- a/system_test/broker_failure/config/mirror_producer3.properties +++ b/system_test/broker_failure/config/mirror_producer3.properties @@ -24,5 +24,5 @@ zk.connectiontimeout.ms=1000000 producer.type=async # to avoid dropping events if the queue is full, wait indefinitely -queue.enqueueTimeout.ms=-1 +queue.enqueue.timeout.ms=-1 diff --git a/system_test/migration_tool_testsuite/config/migration_producer.properties b/system_test/migration_tool_testsuite/config/migration_producer.properties index a923ee3..e3bbc25 100644 --- a/system_test/migration_tool_testsuite/config/migration_producer.properties +++ b/system_test/migration_tool_testsuite/config/migration_producer.properties @@ -61,7 +61,7 @@ serializer.class=kafka.serializer.DefaultEncoder # 0: events will be enqueued immediately or dropped if the queue is full # -ve: enqueue will block indefinitely if the queue is full # +ve: enqueue will block up to this many milliseconds if the queue is full -#queue.enqueueTimeout.ms= +#queue.enqueue.timeout.ms= # the number of messages batched at the producer #batch.size= diff --git a/system_test/mirror_maker/config/mirror_producer.properties b/system_test/mirror_maker/config/mirror_producer.properties index b74c631..ecb3338 100644 --- a/system_test/mirror_maker/config/mirror_producer.properties +++ b/system_test/mirror_maker/config/mirror_producer.properties @@ -24,7 +24,7 @@ zk.connect=localhost:2183 producer.type=async # to avoid dropping events if the queue is full, wait indefinitely -queue.enqueueTimeout.ms=-1 +queue.enqueue.timeout.ms=-1 num.producers.per.broker=2 diff --git a/system_test/mirror_maker_testsuite/config/mirror_producer.properties b/system_test/mirror_maker_testsuite/config/mirror_producer.properties index 3bb5a7b..7db5bfc 100644 --- a/system_test/mirror_maker_testsuite/config/mirror_producer.properties +++ b/system_test/mirror_maker_testsuite/config/mirror_producer.properties @@ -1,5 +1,5 @@ producer.type=async -queue.enqueueTimeout.ms=-1 +queue.enqueue.timeout.ms=-1 broker.list=localhost:9094 compression.codec=0 -- 1.7.1 From 9290b3179f9b14b93bcdd0fa1a78958fe48a100c Mon Sep 17 00:00:00 2001 From: Sriram Subramanian Date: Mon, 7 Jan 2013 17:19:28 -0800 Subject: [PATCH 08/10] Kafka-648 Use uniform conventions for naming properties keys modified: config/server.properties modified: core/src/main/scala/kafka/client/ClientUtils.scala modified: core/src/main/scala/kafka/consumer/ConsoleConsumer.scala modified: core/src/main/scala/kafka/tools/KafkaMigrationTool.java modified: core/src/main/scala/kafka/tools/ReplayLogProducer.scala modified: core/src/test/scala/other/kafka/TestEndToEndLatency.scala modified: core/src/test/scala/other/kafka/TestZKConsumerOffsets.scala modified: core/src/test/scala/unit/kafka/integration/AutoOffsetResetTest.scala modified: core/src/test/scala/unit/kafka/integration/ProducerConsumerTestHarness.scala modified: core/src/test/scala/unit/kafka/utils/TestUtils.scala modified: examples/src/main/java/kafka/examples/Consumer.java modified: perf/src/main/scala/kafka/perf/ConsumerPerformance.scala modified: perf/src/main/scala/kafka/perf/ProducerPerformance.scala modified: system_test/broker_failure/config/server_source1.properties modified: system_test/broker_failure/config/server_source2.properties modified: system_test/broker_failure/config/server_source3.properties modified: system_test/broker_failure/config/server_source4.properties modified: system_test/broker_failure/config/server_target1.properties modified: system_test/broker_failure/config/server_target2.properties modified: system_test/broker_failure/config/server_target3.properties modified: system_test/broker_failure/config/whitelisttest.consumer.properties modified: system_test/migration_tool_testsuite/config/migration_consumer.properties modified: system_test/migration_tool_testsuite/config/server.properties modified: system_test/mirror_maker/config/blacklisttest.consumer.properties modified: system_test/mirror_maker/config/server_source_1_1.properties modified: system_test/mirror_maker/config/server_source_1_2.properties modified: system_test/mirror_maker/config/server_source_2_1.properties modified: system_test/mirror_maker/config/server_source_2_2.properties modified: system_test/mirror_maker/config/server_target_1_1.properties modified: system_test/mirror_maker/config/server_target_1_2.properties modified: system_test/mirror_maker/config/whitelisttest_1.consumer.properties modified: system_test/mirror_maker/config/whitelisttest_2.consumer.properties modified: system_test/mirror_maker_testsuite/config/mirror_consumer.properties modified: system_test/mirror_maker_testsuite/config/server.properties modified: system_test/producer_perf/config/server.properties modified: system_test/replication_testsuite/config/server.properties --- config/server.properties | 6 +++--- core/src/main/scala/kafka/client/ClientUtils.scala | 2 +- .../scala/kafka/consumer/ConsoleConsumer.scala | 10 +++++----- .../main/scala/kafka/tools/KafkaMigrationTool.java | 4 ++-- .../main/scala/kafka/tools/ReplayLogProducer.scala | 4 ++-- .../scala/other/kafka/TestEndToEndLatency.scala | 4 ++-- .../scala/other/kafka/TestZKConsumerOffsets.scala | 2 +- .../kafka/integration/AutoOffsetResetTest.scala | 2 +- .../integration/ProducerConsumerTestHarness.scala | 2 +- .../test/scala/unit/kafka/utils/TestUtils.scala | 4 ++-- .../src/main/java/kafka/examples/Consumer.java | 4 ++-- .../scala/kafka/perf/ConsumerPerformance.scala | 4 ++-- .../scala/kafka/perf/ProducerPerformance.scala | 2 +- .../config/server_source1.properties | 8 ++++---- .../config/server_source2.properties | 8 ++++---- .../config/server_source3.properties | 8 ++++---- .../config/server_source4.properties | 8 ++++---- .../config/server_target1.properties | 8 ++++---- .../config/server_target2.properties | 8 ++++---- .../config/server_target3.properties | 8 ++++---- .../config/whitelisttest.consumer.properties | 4 ++-- .../config/migration_consumer.properties | 2 +- .../config/server.properties | 16 ++++++++-------- .../config/blacklisttest.consumer.properties | 4 ++-- .../config/server_source_1_1.properties | 8 ++++---- .../config/server_source_1_2.properties | 8 ++++---- .../config/server_source_2_1.properties | 8 ++++---- .../config/server_source_2_2.properties | 8 ++++---- .../config/server_target_1_1.properties | 8 ++++---- .../config/server_target_1_2.properties | 8 ++++---- .../config/whitelisttest_1.consumer.properties | 4 ++-- .../config/whitelisttest_2.consumer.properties | 4 ++-- .../config/mirror_consumer.properties | 12 ++++++------ .../config/server.properties | 14 +++++++------- system_test/producer_perf/config/server.properties | 8 ++++---- .../replication_testsuite/config/server.properties | 14 +++++++------- 36 files changed, 118 insertions(+), 118 deletions(-) diff --git a/config/server.properties b/config/server.properties index 241e46e..3af9dc1 100644 --- a/config/server.properties +++ b/config/server.properties @@ -30,10 +30,10 @@ port=9092 #host.name=localhost # The number of threads handling network requests -network.threads=2 +num.network.threads=2 # The number of threads doing disk I/O -io.threads=2 +num.io.threads=2 # The send buffer (SO_SNDBUF) used by the socket server socket.send.buffer=1048576 @@ -75,7 +75,7 @@ log.flush.interval=10000 log.default.flush.interval.ms=1000 # Per-topic overrides for log.default.flush.interval.ms -#topic.flush.intervals.ms=topic1:1000, topic2:3000 +#log.flush.intervals.ms.per.topic=topic1:1000, topic2:3000 # The interval (in ms) at which logs are checked to see if they need to be flushed to disk. log.default.flush.interval.ms=1000 diff --git a/core/src/main/scala/kafka/client/ClientUtils.scala b/core/src/main/scala/kafka/client/ClientUtils.scala index c61833b..7f0c1a1 100644 --- a/core/src/main/scala/kafka/client/ClientUtils.scala +++ b/core/src/main/scala/kafka/client/ClientUtils.scala @@ -58,7 +58,7 @@ object ClientUtils extends Logging{ def fetchTopicMetadata(topics: Set[String], brokers: Seq[Broker], clientId: String): TopicMetadataResponse = { val props = new Properties() props.put("broker.list", brokers.map(_.getConnectionString()).mkString(",")) - props.put("clientid", clientId) + props.put("client.id", clientId) val producerConfig = new ProducerConfig(props) fetchTopicMetadata(topics, brokers, producerConfig) } diff --git a/core/src/main/scala/kafka/consumer/ConsoleConsumer.scala b/core/src/main/scala/kafka/consumer/ConsoleConsumer.scala index 9378357..4aa54dc 100644 --- a/core/src/main/scala/kafka/consumer/ConsoleConsumer.scala +++ b/core/src/main/scala/kafka/consumer/ConsoleConsumer.scala @@ -144,14 +144,14 @@ object ConsoleConsumer extends Logging { } val props = new Properties() - props.put("groupid", options.valueOf(groupIdOpt)) - props.put("socket.buffersize", options.valueOf(socketBufferSizeOpt).toString) + props.put("group.id", options.valueOf(groupIdOpt)) + props.put("socket.buffer.size", options.valueOf(socketBufferSizeOpt).toString) props.put("fetch.size", options.valueOf(fetchSizeOpt).toString) props.put("min.fetch.bytes", options.valueOf(minFetchBytesOpt).toString) props.put("max.fetch.wait.ms", options.valueOf(maxWaitMsOpt).toString) - props.put("autocommit.enable", "true") - props.put("autocommit.interval.ms", options.valueOf(autoCommitIntervalOpt).toString) - props.put("autooffset.reset", if(options.has(resetBeginningOpt)) "smallest" else "largest") + props.put("auto.commit.enable", "true") + props.put("auto.commit.interval.ms", options.valueOf(autoCommitIntervalOpt).toString) + props.put("auto.offset.reset", if(options.has(resetBeginningOpt)) "smallest" else "largest") props.put("zk.connect", options.valueOf(zkConnectOpt)) props.put("consumer.timeout.ms", options.valueOf(consumerTimeoutMsOpt).toString) val config = new ConsumerConfig(props) diff --git a/core/src/main/scala/kafka/tools/KafkaMigrationTool.java b/core/src/main/scala/kafka/tools/KafkaMigrationTool.java index 36a119b..1f5c7ba 100644 --- a/core/src/main/scala/kafka/tools/KafkaMigrationTool.java +++ b/core/src/main/scala/kafka/tools/KafkaMigrationTool.java @@ -182,9 +182,9 @@ public class KafkaMigrationTool Properties kafkaConsumerProperties_07 = new Properties(); kafkaConsumerProperties_07.load(new FileInputStream(consumerConfigFile_07)); /** Disable shallow iteration because the message format is different between 07 and 08, we have to get each individual message **/ - if(kafkaConsumerProperties_07.getProperty("shallowiterator.enable", "").equals("true")){ + if(kafkaConsumerProperties_07.getProperty("shallow.iterator.enable", "").equals("true")){ logger.warn("Shallow iterator should not be used in the migration tool"); - kafkaConsumerProperties_07.setProperty("shallowiterator.enable", "false"); + kafkaConsumerProperties_07.setProperty("shallow.iterator.enable", "false"); } Object consumerConfig_07 = ConsumerConfigConstructor_07.newInstance(kafkaConsumerProperties_07); diff --git a/core/src/main/scala/kafka/tools/ReplayLogProducer.scala b/core/src/main/scala/kafka/tools/ReplayLogProducer.scala index 4cf0a73..8dff2b8 100644 --- a/core/src/main/scala/kafka/tools/ReplayLogProducer.scala +++ b/core/src/main/scala/kafka/tools/ReplayLogProducer.scala @@ -42,10 +42,10 @@ object ReplayLogProducer extends Logging { // consumer properties val consumerProps = new Properties - consumerProps.put("groupid", GroupId) + consumerProps.put("group.id", GroupId) consumerProps.put("zk.connect", config.zkConnect) consumerProps.put("consumer.timeout.ms", "10000") - consumerProps.put("autooffset.reset", OffsetRequest.SmallestTimeString) + consumerProps.put("auto.offset.reset", OffsetRequest.SmallestTimeString) consumerProps.put("fetch.size", (1024*1024).toString) consumerProps.put("socket.buffer.size", (2 * 1024 * 1024).toString) val consumerConfig = new ConsumerConfig(consumerProps) diff --git a/core/src/test/scala/other/kafka/TestEndToEndLatency.scala b/core/src/test/scala/other/kafka/TestEndToEndLatency.scala index 5be4f4e..98c12b7 100644 --- a/core/src/test/scala/other/kafka/TestEndToEndLatency.scala +++ b/core/src/test/scala/other/kafka/TestEndToEndLatency.scala @@ -35,9 +35,9 @@ object TestEndToEndLatency { val topic = "test" val consumerProps = new Properties() - consumerProps.put("groupid", topic) + consumerProps.put("group.id", topic) consumerProps.put("auto.commit", "true") - consumerProps.put("autooffset.reset", "largest") + consumerProps.put("auto.offset.reset", "largest") consumerProps.put("zk.connect", zkConnect) consumerProps.put("socket.timeout.ms", 1201000.toString) diff --git a/core/src/test/scala/other/kafka/TestZKConsumerOffsets.scala b/core/src/test/scala/other/kafka/TestZKConsumerOffsets.scala index 5b72eed..31534ca 100644 --- a/core/src/test/scala/other/kafka/TestZKConsumerOffsets.scala +++ b/core/src/test/scala/other/kafka/TestZKConsumerOffsets.scala @@ -31,7 +31,7 @@ object TestZKConsumerOffsets { val topic = args(1) val autoOffsetReset = args(2) val props = Utils.loadProps(args(0)) - props.put("autooffset.reset", "largest") + props.put("auto.offset.reset", "largest") val config = new ConsumerConfig(props) val consumerConnector: ConsumerConnector = Consumer.create(config) diff --git a/core/src/test/scala/unit/kafka/integration/AutoOffsetResetTest.scala b/core/src/test/scala/unit/kafka/integration/AutoOffsetResetTest.scala index d7945a5..36ae9ae 100644 --- a/core/src/test/scala/unit/kafka/integration/AutoOffsetResetTest.scala +++ b/core/src/test/scala/unit/kafka/integration/AutoOffsetResetTest.scala @@ -78,7 +78,7 @@ class AutoOffsetResetTest extends JUnit3Suite with KafkaServerTestHarness with L // update offset in zookeeper for consumer to jump "forward" in time val dirs = new ZKGroupTopicDirs(group, topic) var consumerProps = TestUtils.createConsumerProperties(zkConnect, group, testConsumer) - consumerProps.put("autooffset.reset", resetTo) + consumerProps.put("auto.offset.reset", resetTo) consumerProps.put("consumer.timeout.ms", "2000") consumerProps.put("max.fetch.wait.ms", "0") val consumerConfig = new ConsumerConfig(consumerProps) diff --git a/core/src/test/scala/unit/kafka/integration/ProducerConsumerTestHarness.scala b/core/src/test/scala/unit/kafka/integration/ProducerConsumerTestHarness.scala index caea858..f89438d 100644 --- a/core/src/test/scala/unit/kafka/integration/ProducerConsumerTestHarness.scala +++ b/core/src/test/scala/unit/kafka/integration/ProducerConsumerTestHarness.scala @@ -39,7 +39,7 @@ trait ProducerConsumerTestHarness extends JUnit3Suite with KafkaServerTestHarnes props.put("connect.timeout.ms", "100000") props.put("reconnect.interval", "10000") props.put("producer.retry.backoff.ms", "1000") - props.put("producer.num.retries", "3") + props.put("producer.retry.count", "3") props.put("producer.request.required.acks", "-1") props.put("serializer.class", classOf[StringEncoder].getName.toString) producer = new Producer(new ProducerConfig(props)) diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index dec245a..b7cccdb 100644 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -145,8 +145,8 @@ object TestUtils extends Logging { props.put("consumer.timeout.ms", consumerTimeout.toString) props.put("zk.sessiontimeout.ms", "400") props.put("zk.synctime.ms", "200") - props.put("autocommit.interval.ms", "1000") - props.put("rebalance.retries.max", "4") + props.put("auto.commit.interval.ms", "1000") + props.put("max.rebalance.retries", "4") props } diff --git a/examples/src/main/java/kafka/examples/Consumer.java b/examples/src/main/java/kafka/examples/Consumer.java index 2b87560..e8054ca 100644 --- a/examples/src/main/java/kafka/examples/Consumer.java +++ b/examples/src/main/java/kafka/examples/Consumer.java @@ -44,10 +44,10 @@ public class Consumer extends Thread { Properties props = new Properties(); props.put("zk.connect", KafkaProperties.zkConnect); - props.put("groupid", KafkaProperties.groupId); + props.put("group.id", KafkaProperties.groupId); props.put("zk.sessiontimeout.ms", "400"); props.put("zk.synctime.ms", "200"); - props.put("autocommit.interval.ms", "1000"); + props.put("auto.commit.interval.ms", "1000"); return new ConsumerConfig(props); diff --git a/perf/src/main/scala/kafka/perf/ConsumerPerformance.scala b/perf/src/main/scala/kafka/perf/ConsumerPerformance.scala index a720ced..7dd6165 100644 --- a/perf/src/main/scala/kafka/perf/ConsumerPerformance.scala +++ b/perf/src/main/scala/kafka/perf/ConsumerPerformance.scala @@ -124,10 +124,10 @@ object ConsumerPerformance { } val props = new Properties - props.put("groupid", options.valueOf(groupIdOpt)) + props.put("group.id", options.valueOf(groupIdOpt)) props.put("socket.buffer.size", options.valueOf(socketBufferSizeOpt).toString) props.put("fetch.size", options.valueOf(fetchSizeOpt).toString) - props.put("autooffset.reset", if(options.has(resetBeginningOffsetOpt)) "largest" else "smallest") + props.put("auto.offset.reset", if(options.has(resetBeginningOffsetOpt)) "largest" else "smallest") props.put("zk.connect", options.valueOf(zkConnectOpt)) props.put("consumer.timeout.ms", "5000") val consumerConfig = new ConsumerConfig(props) diff --git a/perf/src/main/scala/kafka/perf/ProducerPerformance.scala b/perf/src/main/scala/kafka/perf/ProducerPerformance.scala index 1942a38..a55e1b4 100644 --- a/perf/src/main/scala/kafka/perf/ProducerPerformance.scala +++ b/perf/src/main/scala/kafka/perf/ProducerPerformance.scala @@ -199,7 +199,7 @@ object ProducerPerformance extends Logging { } props.put("producer.request.required.acks", config.producerRequestRequiredAcks.toString) props.put("producer.request.timeout.ms", config.producerRequestTimeoutMs.toString) - props.put("producer.num.retries", config.producerNumRetries.toString) + props.put("producer.retry.count", config.producerNumRetries.toString) props.put("producer.retry.backoff.ms", config.producerRetryBackoffMs.toString) props.put("serializer.class", classOf[DefaultEncoder].getName.toString) props.put("key.serializer.class", classOf[NullEncoder[Long]].getName.toString) diff --git a/system_test/broker_failure/config/server_source1.properties b/system_test/broker_failure/config/server_source1.properties index 1a16c2c..34c4c15 100644 --- a/system_test/broker_failure/config/server_source1.properties +++ b/system_test/broker_failure/config/server_source1.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=1 +broker.id=1 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=10000000 +log.segment.size=10000000 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2181 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/broker_failure/config/server_source2.properties b/system_test/broker_failure/config/server_source2.properties index 032bbcc..358f88f 100644 --- a/system_test/broker_failure/config/server_source2.properties +++ b/system_test/broker_failure/config/server_source2.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=2 +broker.id=2 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=10000000 +log.segment.size=10000000 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2181 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/broker_failure/config/server_source3.properties b/system_test/broker_failure/config/server_source3.properties index 05b3a97..360060a 100644 --- a/system_test/broker_failure/config/server_source3.properties +++ b/system_test/broker_failure/config/server_source3.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=3 +broker.id=3 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=10000000 +log.segment.size=10000000 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2181 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/broker_failure/config/server_source4.properties b/system_test/broker_failure/config/server_source4.properties index c94204d..485c32c 100644 --- a/system_test/broker_failure/config/server_source4.properties +++ b/system_test/broker_failure/config/server_source4.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=4 +broker.id=4 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=10000000 +log.segment.size=10000000 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2181 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/broker_failure/config/server_target1.properties b/system_test/broker_failure/config/server_target1.properties index e9cc038..6f3f0e1 100644 --- a/system_test/broker_failure/config/server_target1.properties +++ b/system_test/broker_failure/config/server_target1.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=1 +broker.id=1 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=10000000 +log.segment.size=10000000 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2182 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/broker_failure/config/server_target2.properties b/system_test/broker_failure/config/server_target2.properties index 6007fa8..3b5c1a2 100644 --- a/system_test/broker_failure/config/server_target2.properties +++ b/system_test/broker_failure/config/server_target2.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=2 +broker.id=2 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=10000000 +log.segment.size=10000000 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2182 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/broker_failure/config/server_target3.properties b/system_test/broker_failure/config/server_target3.properties index 9ac0b06..bc8efb7 100644 --- a/system_test/broker_failure/config/server_target3.properties +++ b/system_test/broker_failure/config/server_target3.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=3 +broker.id=3 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=10000000 +log.segment.size=10000000 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2182 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/broker_failure/config/whitelisttest.consumer.properties b/system_test/broker_failure/config/whitelisttest.consumer.properties index aaa3f7c..4c1ad81 100644 --- a/system_test/broker_failure/config/whitelisttest.consumer.properties +++ b/system_test/broker_failure/config/whitelisttest.consumer.properties @@ -23,7 +23,7 @@ zk.connect=localhost:2181 zk.connectiontimeout.ms=1000000 #consumer group id -groupid=group1 +group.id=group1 mirror.topics.whitelist=test_1,test_2 -autooffset.reset=smallest +auto.offset.reset=smallest diff --git a/system_test/migration_tool_testsuite/config/migration_consumer.properties b/system_test/migration_tool_testsuite/config/migration_consumer.properties index 184f1de..42ed98d 100644 --- a/system_test/migration_tool_testsuite/config/migration_consumer.properties +++ b/system_test/migration_tool_testsuite/config/migration_consumer.properties @@ -23,7 +23,7 @@ zk.connect=127.0.0.1:2188 zk.connectiontimeout.ms=1000000 #consumer group id -groupid=test-consumer-group +group.id=test-consumer-group #consumer timeout #consumer.timeout.ms=5000 diff --git a/system_test/migration_tool_testsuite/config/server.properties b/system_test/migration_tool_testsuite/config/server.properties index 6976869..000d8a0 100644 --- a/system_test/migration_tool_testsuite/config/server.properties +++ b/system_test/migration_tool_testsuite/config/server.properties @@ -17,12 +17,12 @@ ############################# Server Basics ############################# # The id of the broker. This must be set to a unique integer for each broker. -brokerid=0 +broker.id=0 # Hostname the broker will advertise to consumers. If not set, kafka will use the value returned # from InetAddress.getLocalHost(). If there are multiple interfaces getLocalHost # may not be what you want. -#hostname= +#host.name= ############################# Socket Server Settings ############################# @@ -31,10 +31,10 @@ brokerid=0 port=9091 # The number of threads handling network requests -network.threads=2 +num.network.threads=2 # The number of threads doing disk I/O -io.threads=2 +num.io.threads=2 # The send buffer (SO_SNDBUF) used by the socket server socket.send.buffer=1048576 @@ -76,7 +76,7 @@ log.flush.interval=10000 log.default.flush.interval.ms=1000 # Per-topic overrides for log.default.flush.interval.ms -#topic.flush.intervals.ms=topic1:1000, topic2:3000 +#log.flush.intervals.ms.per.topic=topic1:1000, topic2:3000 # The interval (in ms) at which logs are checked to see if they need to be flushed to disk. log.default.flush.scheduler.interval.ms=1000 @@ -96,9 +96,9 @@ log.retention.hours=168 #log.retention.size=1073741824 # The maximum size of a log segment file. When this size is reached a new log segment will be created. -#log.file.size=536870912 -#log.file.size=102400 -log.file.size=128 +#log.segment.size=536870912 +#log.segment.size=102400 +log.segment.size=128 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies diff --git a/system_test/mirror_maker/config/blacklisttest.consumer.properties b/system_test/mirror_maker/config/blacklisttest.consumer.properties index 6ea85ec..404bf0d 100644 --- a/system_test/mirror_maker/config/blacklisttest.consumer.properties +++ b/system_test/mirror_maker/config/blacklisttest.consumer.properties @@ -23,6 +23,6 @@ zk.connect=localhost:2181 zk.connectiontimeout.ms=1000000 #consumer group id -groupid=group1 -shallowiterator.enable=true +group.id=group1 +shallow.iterator.enable=true diff --git a/system_test/mirror_maker/config/server_source_1_1.properties b/system_test/mirror_maker/config/server_source_1_1.properties index d89c4fb..cc6ca98 100644 --- a/system_test/mirror_maker/config/server_source_1_1.properties +++ b/system_test/mirror_maker/config/server_source_1_1.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=1 +broker.id=1 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=10000000 +log.segment.size=10000000 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2181 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/mirror_maker/config/server_source_1_2.properties b/system_test/mirror_maker/config/server_source_1_2.properties index 063d68b..89ac6df 100644 --- a/system_test/mirror_maker/config/server_source_1_2.properties +++ b/system_test/mirror_maker/config/server_source_1_2.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=2 +broker.id=2 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=536870912 +log.segment.size=536870912 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2181 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/mirror_maker/config/server_source_2_1.properties b/system_test/mirror_maker/config/server_source_2_1.properties index 998b460..f1ee6f2 100644 --- a/system_test/mirror_maker/config/server_source_2_1.properties +++ b/system_test/mirror_maker/config/server_source_2_1.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=1 +broker.id=1 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=536870912 +log.segment.size=536870912 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2182 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/mirror_maker/config/server_source_2_2.properties b/system_test/mirror_maker/config/server_source_2_2.properties index 81427ae..cd015fa 100644 --- a/system_test/mirror_maker/config/server_source_2_2.properties +++ b/system_test/mirror_maker/config/server_source_2_2.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=2 +broker.id=2 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=536870912 +log.segment.size=536870912 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2182 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/mirror_maker/config/server_target_1_1.properties b/system_test/mirror_maker/config/server_target_1_1.properties index 0265f4e..70b17af 100644 --- a/system_test/mirror_maker/config/server_target_1_1.properties +++ b/system_test/mirror_maker/config/server_target_1_1.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=1 +broker.id=1 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=536870912 +log.segment.size=536870912 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2183 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/mirror_maker/config/server_target_1_2.properties b/system_test/mirror_maker/config/server_target_1_2.properties index a31e9ca..3fef21e 100644 --- a/system_test/mirror_maker/config/server_target_1_2.properties +++ b/system_test/mirror_maker/config/server_target_1_2.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=2 +broker.id=2 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=536870912 +log.segment.size=536870912 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2183 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/mirror_maker/config/whitelisttest_1.consumer.properties b/system_test/mirror_maker/config/whitelisttest_1.consumer.properties index 6ea85ec..404bf0d 100644 --- a/system_test/mirror_maker/config/whitelisttest_1.consumer.properties +++ b/system_test/mirror_maker/config/whitelisttest_1.consumer.properties @@ -23,6 +23,6 @@ zk.connect=localhost:2181 zk.connectiontimeout.ms=1000000 #consumer group id -groupid=group1 -shallowiterator.enable=true +group.id=group1 +shallow.iterator.enable=true diff --git a/system_test/mirror_maker/config/whitelisttest_2.consumer.properties b/system_test/mirror_maker/config/whitelisttest_2.consumer.properties index e11112f..fd2ab57 100644 --- a/system_test/mirror_maker/config/whitelisttest_2.consumer.properties +++ b/system_test/mirror_maker/config/whitelisttest_2.consumer.properties @@ -23,6 +23,6 @@ zk.connect=localhost:2182 zk.connectiontimeout.ms=1000000 #consumer group id -groupid=group1 -shallowiterator.enable=true +group.id=group1 +shallow.iterator.enable=true diff --git a/system_test/mirror_maker_testsuite/config/mirror_consumer.properties b/system_test/mirror_maker_testsuite/config/mirror_consumer.properties index 6517976..cb68040 100644 --- a/system_test/mirror_maker_testsuite/config/mirror_consumer.properties +++ b/system_test/mirror_maker_testsuite/config/mirror_consumer.properties @@ -1,12 +1,12 @@ zk.connect=localhost:2108 zk.connectiontimeout.ms=1000000 -groupid=mm_regtest_grp -autocommit.interval.ms=120000 -autooffset.reset=smallest +group.id=mm_regtest_grp +auto.commit.interval.ms=120000 +auto.offset.reset=smallest #fetch.size=1048576 -#rebalance.retries.max=4 +#max.rebalance.retries=4 #rebalance.backoff.ms=2000 -socket.buffersize=1048576 +socket.buffer.size=1048576 fetch.size=1048576 zk.synctime.ms=15000 -shallowiterator.enable=true +shallow.iterator.enable=true diff --git a/system_test/mirror_maker_testsuite/config/server.properties b/system_test/mirror_maker_testsuite/config/server.properties index 8ef65ba..ca0a85e 100644 --- a/system_test/mirror_maker_testsuite/config/server.properties +++ b/system_test/mirror_maker_testsuite/config/server.properties @@ -17,12 +17,12 @@ ############################# Server Basics ############################# # The id of the broker. This must be set to a unique integer for each broker. -brokerid=0 +broker.id=0 # Hostname the broker will advertise to consumers. If not set, kafka will use the value returned # from InetAddress.getLocalHost(). If there are multiple interfaces getLocalHost # may not be what you want. -#hostname= +#host.name= ############################# Socket Server Settings ############################# @@ -31,10 +31,10 @@ brokerid=0 port=9091 # The number of threads handling network requests -network.threads=2 +num.network.threads=2 # The number of threads doing disk I/O -io.threads=2 +num.io.threads=2 # The send buffer (SO_SNDBUF) used by the socket server socket.send.buffer=1048576 @@ -76,7 +76,7 @@ log.flush.interval=10000 log.default.flush.interval.ms=1000 # Per-topic overrides for log.default.flush.interval.ms -#topic.flush.intervals.ms=topic1:1000, topic2:3000 +#log.flush.intervals.ms.per.topic=topic1:1000, topic2:3000 # The interval (in ms) at which logs are checked to see if they need to be flushed to disk. log.default.flush.scheduler.interval.ms=1000 @@ -97,8 +97,8 @@ log.retention.hours=168 log.retention.size=-1 # The maximum size of a log segment file. When this size is reached a new log segment will be created. -#log.file.size=536870912 -log.file.size=102400 +#log.segment.size=536870912 +log.segment.size=102400 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies diff --git a/system_test/producer_perf/config/server.properties b/system_test/producer_perf/config/server.properties index abd0765..55c54ea 100644 --- a/system_test/producer_perf/config/server.properties +++ b/system_test/producer_perf/config/server.properties @@ -15,12 +15,12 @@ # see kafka.server.KafkaConfig for additional details and defaults # the id of the broker -brokerid=0 +broker.id=0 # hostname of broker. If not set, will pick up from the value returned # from getLocalHost. If there are multiple interfaces getLocalHost # may not be what you want. -# hostname= +# host.name= # number of logical partitions on this broker num.partitions=1 @@ -41,7 +41,7 @@ socket.send.buffer=1048576 socket.receive.buffer=1048576 # the maximum size of a log segment -log.file.size=536870912 +log.segment.size=536870912 # the interval between running cleanup on the logs log.cleanup.interval.mins=1 @@ -66,7 +66,7 @@ zk.connect=localhost:2181 zk.connectiontimeout.ms=1000000 # time based topic flush intervals in ms -#topic.flush.intervals.ms=topic:1000 +#log.flush.intervals.ms.per.topic=topic:1000 # default time based flush interval in ms log.default.flush.interval.ms=1000 diff --git a/system_test/replication_testsuite/config/server.properties b/system_test/replication_testsuite/config/server.properties index 8ef65ba..ca0a85e 100644 --- a/system_test/replication_testsuite/config/server.properties +++ b/system_test/replication_testsuite/config/server.properties @@ -17,12 +17,12 @@ ############################# Server Basics ############################# # The id of the broker. This must be set to a unique integer for each broker. -brokerid=0 +broker.id=0 # Hostname the broker will advertise to consumers. If not set, kafka will use the value returned # from InetAddress.getLocalHost(). If there are multiple interfaces getLocalHost # may not be what you want. -#hostname= +#host.name= ############################# Socket Server Settings ############################# @@ -31,10 +31,10 @@ brokerid=0 port=9091 # The number of threads handling network requests -network.threads=2 +num.network.threads=2 # The number of threads doing disk I/O -io.threads=2 +num.io.threads=2 # The send buffer (SO_SNDBUF) used by the socket server socket.send.buffer=1048576 @@ -76,7 +76,7 @@ log.flush.interval=10000 log.default.flush.interval.ms=1000 # Per-topic overrides for log.default.flush.interval.ms -#topic.flush.intervals.ms=topic1:1000, topic2:3000 +#log.flush.intervals.ms.per.topic=topic1:1000, topic2:3000 # The interval (in ms) at which logs are checked to see if they need to be flushed to disk. log.default.flush.scheduler.interval.ms=1000 @@ -97,8 +97,8 @@ log.retention.hours=168 log.retention.size=-1 # The maximum size of a log segment file. When this size is reached a new log segment will be created. -#log.file.size=536870912 -log.file.size=102400 +#log.segment.size=536870912 +log.segment.size=102400 # The interval at which log segments are checked to see if they can be deleted according # to the retention policies -- 1.7.1 From e439ff0ab8511a1ec1a0a80d85bef62c03debbe7 Mon Sep 17 00:00:00 2001 From: Sriram Subramanian Date: Tue, 8 Jan 2013 14:58:58 -0800 Subject: [PATCH 09/10] Kafka 648 - Use uniform convention for naming properties keys modified: config/producer.properties modified: config/server.properties modified: core/src/main/scala/kafka/server/KafkaConfig.scala modified: core/src/main/scala/kafka/server/ReplicaFetcherThread.scala modified: core/src/main/scala/kafka/utils/ZkUtils.scala modified: core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala modified: core/src/test/scala/unit/kafka/utils/TestUtils.scala modified: examples/src/main/java/kafka/examples/Consumer.java modified: system_test/broker_failure/config/mirror_producer.properties modified: system_test/broker_failure/config/mirror_producer1.properties modified: system_test/broker_failure/config/mirror_producer2.properties modified: system_test/broker_failure/config/mirror_producer3.properties modified: system_test/broker_failure/config/server_source1.properties modified: system_test/broker_failure/config/server_source2.properties modified: system_test/broker_failure/config/server_source3.properties modified: system_test/broker_failure/config/server_source4.properties modified: system_test/broker_failure/config/server_target1.properties modified: system_test/broker_failure/config/server_target2.properties modified: system_test/broker_failure/config/server_target3.properties modified: system_test/broker_failure/config/whitelisttest.consumer.properties modified: system_test/migration_tool_testsuite/config/migration_consumer.properties modified: system_test/migration_tool_testsuite/config/migration_producer.properties modified: system_test/migration_tool_testsuite/config/server.properties modified: system_test/mirror_maker/config/blacklisttest.consumer.properties modified: system_test/mirror_maker/config/mirror_producer.properties modified: system_test/mirror_maker/config/server_source_1_1.properties modified: system_test/mirror_maker/config/server_source_1_2.properties modified: system_test/mirror_maker/config/server_source_2_1.properties modified: system_test/mirror_maker/config/server_source_2_2.properties modified: system_test/mirror_maker/config/server_target_1_1.properties modified: system_test/mirror_maker/config/server_target_1_2.properties modified: system_test/mirror_maker/config/whitelisttest_1.consumer.properties modified: system_test/mirror_maker/config/whitelisttest_2.consumer.properties modified: system_test/mirror_maker_testsuite/config/mirror_consumer.properties modified: system_test/mirror_maker_testsuite/config/server.properties modified: system_test/producer_perf/config/server.properties modified: system_test/replication_testsuite/config/server.properties --- config/producer.properties | 13 ------------- config/server.properties | 3 --- core/src/main/scala/kafka/server/KafkaConfig.scala | 4 ++-- .../scala/kafka/server/ReplicaFetcherThread.scala | 4 ++-- core/src/main/scala/kafka/utils/ZkUtils.scala | 6 +++--- .../scala/unit/kafka/server/LogRecoveryTest.scala | 2 +- .../test/scala/unit/kafka/utils/TestUtils.scala | 4 ++-- .../src/main/java/kafka/examples/Consumer.java | 4 ++-- .../config/mirror_producer.properties | 2 +- .../config/mirror_producer1.properties | 2 +- .../config/mirror_producer2.properties | 2 +- .../config/mirror_producer3.properties | 2 +- .../config/server_source1.properties | 2 +- .../config/server_source2.properties | 2 +- .../config/server_source3.properties | 2 +- .../config/server_source4.properties | 2 +- .../config/server_target1.properties | 2 +- .../config/server_target2.properties | 2 +- .../config/server_target3.properties | 2 +- .../config/whitelisttest.consumer.properties | 2 +- .../config/migration_consumer.properties | 2 +- .../config/migration_producer.properties | 17 ++--------------- .../config/server.properties | 2 +- .../config/blacklisttest.consumer.properties | 2 +- .../mirror_maker/config/mirror_producer.properties | 2 +- .../config/server_source_1_1.properties | 2 +- .../config/server_source_1_2.properties | 2 +- .../config/server_source_2_1.properties | 2 +- .../config/server_source_2_2.properties | 2 +- .../config/server_target_1_1.properties | 2 +- .../config/server_target_1_2.properties | 2 +- .../config/whitelisttest_1.consumer.properties | 2 +- .../config/whitelisttest_2.consumer.properties | 2 +- .../config/mirror_consumer.properties | 4 ++-- .../config/server.properties | 6 +++--- system_test/producer_perf/config/server.properties | 2 +- .../replication_testsuite/config/server.properties | 6 +++--- 37 files changed, 47 insertions(+), 76 deletions(-) diff --git a/config/producer.properties b/config/producer.properties index 650f80c..cb3ec1b 100644 --- a/config/producer.properties +++ b/config/producer.properties @@ -55,16 +55,3 @@ serializer.class=kafka.serializer.StringEncoder # the number of messages batched at the producer #batch.size= - -# the callback handler for one or multiple events -#callback.handler= - -# properties required to initialize the callback handler -#callback.handler.props= - -# the handler for events -#event.handler= - -# properties required to initialize the event handler -#event.handler.props= - diff --git a/config/server.properties b/config/server.properties index 3af9dc1..051694c 100644 --- a/config/server.properties +++ b/config/server.properties @@ -54,9 +54,6 @@ log.dir=/tmp/kafka-logs # for consumption, but also mean more files. num.partitions=1 -# Overrides for for the default given by num.partitions on a per-topic basis -#topic.partition.count.map=topic1:3, topic2:4 - ############################# Log Flush Policy ############################# # The following configurations control the flush of data to disk. This is the most diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index 12592eb..506399c 100644 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -150,10 +150,10 @@ class KafkaConfig private (val props: VerifiableProperties) extends ZKConfig(pro val replicaFetchSize = props.getInt("replica.fetch.size", ConsumerConfig.FetchSize) /* max wait time for each fetcher request issued by follower replicas*/ - val replicaFetchMaxWaitTimeMs = props.getInt("replica.fetch.max.wait.time.ms", 500) + val replicaFetchMaxWaitMs = props.getInt("replica.fetch.max.wait.ms", 500) /* minimum bytes expected for each fetch response. If not enough bytes, wait up to replicaMaxWaitTimeMs */ - val replicaFetchMinExpectedBytes = props.getInt("replica.fetch.min.expected.bytes", 1) + val replicaFetchMinBytes = props.getInt("replica.fetch.min.bytes", 1) /* number of fetcher threads used to replicate messages from a source broker. * Increasing this value can increase the degree of I/O parallelism in the follower broker. */ diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala index 3b855b5..0413f9d 100644 --- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala +++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala @@ -34,8 +34,8 @@ class ReplicaFetcherThread(name:String, socketBufferSize = brokerConfig.replicaSocketBufferSize, fetchSize = brokerConfig.replicaFetchSize, fetcherBrokerId = brokerConfig.brokerId, - maxWait = brokerConfig.replicaFetchMaxWaitTimeMs, - minBytes = brokerConfig.replicaFetchMinExpectedBytes) { + maxWait = brokerConfig.replicaFetchMaxWaitMs, + minBytes = brokerConfig.replicaFetchMinBytes) { // process fetched data def processPartitionData(topicAndPartition: TopicAndPartition, fetchOffset: Long, partitionData: FetchResponsePartitionData) { diff --git a/core/src/main/scala/kafka/utils/ZkUtils.scala b/core/src/main/scala/kafka/utils/ZkUtils.scala index 5ba5938..f594404 100644 --- a/core/src/main/scala/kafka/utils/ZkUtils.scala +++ b/core/src/main/scala/kafka/utils/ZkUtils.scala @@ -785,11 +785,11 @@ class ZKConfig(props: VerifiableProperties) { val zkConnect = props.getString("zk.connect", null) /** zookeeper session timeout */ - val zkSessionTimeoutMs = props.getInt("zk.sessiontimeout.ms", 6000) + val zkSessionTimeoutMs = props.getInt("zk.session.timeout.ms", 6000) /** the max time that the client waits to establish a connection to zookeeper */ - val zkConnectionTimeoutMs = props.getInt("zk.connectiontimeout.ms",zkSessionTimeoutMs) + val zkConnectionTimeoutMs = props.getInt("zk.connection.timeout.ms",zkSessionTimeoutMs) /** how far a ZK follower can be behind a ZK leader */ - val zkSyncTimeMs = props.getInt("zk.synctime.ms", 2000) + val zkSyncTimeMs = props.getInt("zk.sync.time.ms", 2000) } diff --git a/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala b/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala index dfdb0c3..6ebf028 100644 --- a/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala +++ b/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala @@ -30,7 +30,7 @@ class LogRecoveryTest extends JUnit3Suite with ZooKeeperTestHarness { val configs = TestUtils.createBrokerConfigs(2).map(new KafkaConfig(_) { override val replicaMaxLagTimeMs = 5000L override val replicaMaxLagBytes = 10L - override val replicaFetchMinExpectedBytes = 20 + override val replicaFetchMinBytes = 20 }) val topic = "new-topic" val partitionId = 0 diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index b7cccdb..486cda3 100644 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -143,8 +143,8 @@ object TestUtils extends Logging { props.put("group.id", groupId) props.put("consumer.id", consumerId) props.put("consumer.timeout.ms", consumerTimeout.toString) - props.put("zk.sessiontimeout.ms", "400") - props.put("zk.synctime.ms", "200") + props.put("zk.session.timeout.ms", "400") + props.put("zk.sync.time.ms", "200") props.put("auto.commit.interval.ms", "1000") props.put("max.rebalance.retries", "4") diff --git a/examples/src/main/java/kafka/examples/Consumer.java b/examples/src/main/java/kafka/examples/Consumer.java index e8054ca..3460d36 100644 --- a/examples/src/main/java/kafka/examples/Consumer.java +++ b/examples/src/main/java/kafka/examples/Consumer.java @@ -45,8 +45,8 @@ public class Consumer extends Thread Properties props = new Properties(); props.put("zk.connect", KafkaProperties.zkConnect); props.put("group.id", KafkaProperties.groupId); - props.put("zk.sessiontimeout.ms", "400"); - props.put("zk.synctime.ms", "200"); + props.put("zk.session.timeout.ms", "400"); + props.put("zk.sync.time.ms", "200"); props.put("auto.commit.interval.ms", "1000"); return new ConsumerConfig(props); diff --git a/system_test/broker_failure/config/mirror_producer.properties b/system_test/broker_failure/config/mirror_producer.properties index af5d4df..7f80a1e 100644 --- a/system_test/broker_failure/config/mirror_producer.properties +++ b/system_test/broker_failure/config/mirror_producer.properties @@ -18,7 +18,7 @@ zk.connect=localhost:2182 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 producer.type=async diff --git a/system_test/broker_failure/config/mirror_producer1.properties b/system_test/broker_failure/config/mirror_producer1.properties index ac24e7f..81dae76 100644 --- a/system_test/broker_failure/config/mirror_producer1.properties +++ b/system_test/broker_failure/config/mirror_producer1.properties @@ -19,7 +19,7 @@ zk.connect=localhost:2182 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 producer.type=async diff --git a/system_test/broker_failure/config/mirror_producer2.properties b/system_test/broker_failure/config/mirror_producer2.properties index 3c0df5c..714b95d 100644 --- a/system_test/broker_failure/config/mirror_producer2.properties +++ b/system_test/broker_failure/config/mirror_producer2.properties @@ -19,7 +19,7 @@ zk.connect=localhost:2182 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 producer.type=async diff --git a/system_test/broker_failure/config/mirror_producer3.properties b/system_test/broker_failure/config/mirror_producer3.properties index 5f525d3..e8fa72d 100644 --- a/system_test/broker_failure/config/mirror_producer3.properties +++ b/system_test/broker_failure/config/mirror_producer3.properties @@ -19,7 +19,7 @@ zk.connect=localhost:2182 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 producer.type=async diff --git a/system_test/broker_failure/config/server_source1.properties b/system_test/broker_failure/config/server_source1.properties index 34c4c15..d548768 100644 --- a/system_test/broker_failure/config/server_source1.properties +++ b/system_test/broker_failure/config/server_source1.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2181 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/broker_failure/config/server_source2.properties b/system_test/broker_failure/config/server_source2.properties index 358f88f..4d4cead 100644 --- a/system_test/broker_failure/config/server_source2.properties +++ b/system_test/broker_failure/config/server_source2.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2181 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/broker_failure/config/server_source3.properties b/system_test/broker_failure/config/server_source3.properties index 360060a..8ed138c 100644 --- a/system_test/broker_failure/config/server_source3.properties +++ b/system_test/broker_failure/config/server_source3.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2181 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/broker_failure/config/server_source4.properties b/system_test/broker_failure/config/server_source4.properties index 485c32c..af62557 100644 --- a/system_test/broker_failure/config/server_source4.properties +++ b/system_test/broker_failure/config/server_source4.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2181 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/broker_failure/config/server_target1.properties b/system_test/broker_failure/config/server_target1.properties index 6f3f0e1..edecf42 100644 --- a/system_test/broker_failure/config/server_target1.properties +++ b/system_test/broker_failure/config/server_target1.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2182 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/broker_failure/config/server_target2.properties b/system_test/broker_failure/config/server_target2.properties index 3b5c1a2..a107478 100644 --- a/system_test/broker_failure/config/server_target2.properties +++ b/system_test/broker_failure/config/server_target2.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2182 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/broker_failure/config/server_target3.properties b/system_test/broker_failure/config/server_target3.properties index bc8efb7..ac28247 100644 --- a/system_test/broker_failure/config/server_target3.properties +++ b/system_test/broker_failure/config/server_target3.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2182 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/broker_failure/config/whitelisttest.consumer.properties b/system_test/broker_failure/config/whitelisttest.consumer.properties index 4c1ad81..dd91bd3 100644 --- a/system_test/broker_failure/config/whitelisttest.consumer.properties +++ b/system_test/broker_failure/config/whitelisttest.consumer.properties @@ -20,7 +20,7 @@ zk.connect=localhost:2181 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 #consumer group id group.id=group1 diff --git a/system_test/migration_tool_testsuite/config/migration_consumer.properties b/system_test/migration_tool_testsuite/config/migration_consumer.properties index 42ed98d..8458215 100644 --- a/system_test/migration_tool_testsuite/config/migration_consumer.properties +++ b/system_test/migration_tool_testsuite/config/migration_consumer.properties @@ -20,7 +20,7 @@ zk.connect=127.0.0.1:2188 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 #consumer group id group.id=test-consumer-group diff --git a/system_test/migration_tool_testsuite/config/migration_producer.properties b/system_test/migration_tool_testsuite/config/migration_producer.properties index e3bbc25..3c0f28d 100644 --- a/system_test/migration_tool_testsuite/config/migration_producer.properties +++ b/system_test/migration_tool_testsuite/config/migration_producer.properties @@ -26,10 +26,10 @@ broker.list=localhost:9094,localhost:9095,localhost:9096 #zk.connect= # zookeeper session timeout; default is 6000 -#zk.sessiontimeout.ms= +#zk.session.timeout.ms= # the max time that the client waits to establish a connection to zookeeper; default is 6000 -#zk.connectiontimeout.ms +#zk.connection.timeout.ms # name of the partitioner class for partitioning events; default partition spreads data randomly #partitioner.class= @@ -65,16 +65,3 @@ serializer.class=kafka.serializer.DefaultEncoder # the number of messages batched at the producer #batch.size= - -# the callback handler for one or multiple events -#callback.handler= - -# properties required to initialize the callback handler -#callback.handler.props= - -# the handler for events -#event.handler= - -# properties required to initialize the event handler -#event.handler.props= - diff --git a/system_test/migration_tool_testsuite/config/server.properties b/system_test/migration_tool_testsuite/config/server.properties index 000d8a0..56d2ea9 100644 --- a/system_test/migration_tool_testsuite/config/server.properties +++ b/system_test/migration_tool_testsuite/config/server.properties @@ -117,6 +117,6 @@ enable.zookeeper=true zk.connect=localhost:2181 # Timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 monitoring.period.secs=1 diff --git a/system_test/mirror_maker/config/blacklisttest.consumer.properties b/system_test/mirror_maker/config/blacklisttest.consumer.properties index 404bf0d..ff12015 100644 --- a/system_test/mirror_maker/config/blacklisttest.consumer.properties +++ b/system_test/mirror_maker/config/blacklisttest.consumer.properties @@ -20,7 +20,7 @@ zk.connect=localhost:2181 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 #consumer group id group.id=group1 diff --git a/system_test/mirror_maker/config/mirror_producer.properties b/system_test/mirror_maker/config/mirror_producer.properties index ecb3338..aa8be65 100644 --- a/system_test/mirror_maker/config/mirror_producer.properties +++ b/system_test/mirror_maker/config/mirror_producer.properties @@ -19,7 +19,7 @@ zk.connect=localhost:2183 # broker.list=1:localhost:9094,2:localhost:9095 # timeout in ms for connecting to zookeeper -# zk.connectiontimeout.ms=1000000 +# zk.connection.timeout.ms=1000000 producer.type=async diff --git a/system_test/mirror_maker/config/server_source_1_1.properties b/system_test/mirror_maker/config/server_source_1_1.properties index cc6ca98..6991572 100644 --- a/system_test/mirror_maker/config/server_source_1_1.properties +++ b/system_test/mirror_maker/config/server_source_1_1.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2181 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/mirror_maker/config/server_source_1_2.properties b/system_test/mirror_maker/config/server_source_1_2.properties index 89ac6df..26a574e 100644 --- a/system_test/mirror_maker/config/server_source_1_2.properties +++ b/system_test/mirror_maker/config/server_source_1_2.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2181 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/mirror_maker/config/server_source_2_1.properties b/system_test/mirror_maker/config/server_source_2_1.properties index f1ee6f2..9f8a67c 100644 --- a/system_test/mirror_maker/config/server_source_2_1.properties +++ b/system_test/mirror_maker/config/server_source_2_1.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2182 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/mirror_maker/config/server_source_2_2.properties b/system_test/mirror_maker/config/server_source_2_2.properties index cd015fa..b1d9c29 100644 --- a/system_test/mirror_maker/config/server_source_2_2.properties +++ b/system_test/mirror_maker/config/server_source_2_2.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2182 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/mirror_maker/config/server_target_1_1.properties b/system_test/mirror_maker/config/server_target_1_1.properties index 70b17af..82fa319 100644 --- a/system_test/mirror_maker/config/server_target_1_1.properties +++ b/system_test/mirror_maker/config/server_target_1_1.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2183 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/mirror_maker/config/server_target_1_2.properties b/system_test/mirror_maker/config/server_target_1_2.properties index 3fef21e..61651cc 100644 --- a/system_test/mirror_maker/config/server_target_1_2.properties +++ b/system_test/mirror_maker/config/server_target_1_2.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2183 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/mirror_maker/config/whitelisttest_1.consumer.properties b/system_test/mirror_maker/config/whitelisttest_1.consumer.properties index 404bf0d..ff12015 100644 --- a/system_test/mirror_maker/config/whitelisttest_1.consumer.properties +++ b/system_test/mirror_maker/config/whitelisttest_1.consumer.properties @@ -20,7 +20,7 @@ zk.connect=localhost:2181 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 #consumer group id group.id=group1 diff --git a/system_test/mirror_maker/config/whitelisttest_2.consumer.properties b/system_test/mirror_maker/config/whitelisttest_2.consumer.properties index fd2ab57..f1a902b 100644 --- a/system_test/mirror_maker/config/whitelisttest_2.consumer.properties +++ b/system_test/mirror_maker/config/whitelisttest_2.consumer.properties @@ -20,7 +20,7 @@ zk.connect=localhost:2182 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 #consumer group id group.id=group1 diff --git a/system_test/mirror_maker_testsuite/config/mirror_consumer.properties b/system_test/mirror_maker_testsuite/config/mirror_consumer.properties index cb68040..ebebc0a 100644 --- a/system_test/mirror_maker_testsuite/config/mirror_consumer.properties +++ b/system_test/mirror_maker_testsuite/config/mirror_consumer.properties @@ -1,5 +1,5 @@ zk.connect=localhost:2108 -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 group.id=mm_regtest_grp auto.commit.interval.ms=120000 auto.offset.reset=smallest @@ -8,5 +8,5 @@ auto.offset.reset=smallest #rebalance.backoff.ms=2000 socket.buffer.size=1048576 fetch.size=1048576 -zk.synctime.ms=15000 +zk.sync.time.ms=15000 shallow.iterator.enable=true diff --git a/system_test/mirror_maker_testsuite/config/server.properties b/system_test/mirror_maker_testsuite/config/server.properties index ca0a85e..c7ef370 100644 --- a/system_test/mirror_maker_testsuite/config/server.properties +++ b/system_test/mirror_maker_testsuite/config/server.properties @@ -117,7 +117,7 @@ enable.zookeeper=true zk.connect=localhost:2181 # Timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 monitoring.period.secs=1 max.message.size=1000000 @@ -134,6 +134,6 @@ replica.max.lag.bytes=4000 replica.socket.timeout.ms=30000 replica.socket.buffersize=65536 replica.fetch.size=1048576 -replica.fetch.wait.time.ms=500 +replica.fetch.max.wait.ms=500 replica.fetch.min.bytes=4096 -replica.fetchers=1 +num.replica.fetchers=1 diff --git a/system_test/producer_perf/config/server.properties b/system_test/producer_perf/config/server.properties index 55c54ea..56208a7 100644 --- a/system_test/producer_perf/config/server.properties +++ b/system_test/producer_perf/config/server.properties @@ -63,7 +63,7 @@ enable.zookeeper=true zk.connect=localhost:2181 # timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 # time based topic flush intervals in ms #log.flush.intervals.ms.per.topic=topic:1000 diff --git a/system_test/replication_testsuite/config/server.properties b/system_test/replication_testsuite/config/server.properties index ca0a85e..c7ef370 100644 --- a/system_test/replication_testsuite/config/server.properties +++ b/system_test/replication_testsuite/config/server.properties @@ -117,7 +117,7 @@ enable.zookeeper=true zk.connect=localhost:2181 # Timeout in ms for connecting to zookeeper -zk.connectiontimeout.ms=1000000 +zk.connection.timeout.ms=1000000 monitoring.period.secs=1 max.message.size=1000000 @@ -134,6 +134,6 @@ replica.max.lag.bytes=4000 replica.socket.timeout.ms=30000 replica.socket.buffersize=65536 replica.fetch.size=1048576 -replica.fetch.wait.time.ms=500 +replica.fetch.max.wait.ms=500 replica.fetch.min.bytes=4096 -replica.fetchers=1 +num.replica.fetchers=1 -- 1.7.1 From 03e5df0cdee30cbf5c9b9dc4cf93a1ee6eca4b56 Mon Sep 17 00:00:00 2001 From: Sriram Subramanian Date: Wed, 9 Jan 2013 11:45:03 -0800 Subject: [PATCH 10/10] Kafka 648 - Use uniform convention for naming properties keys modified: system_test/migration_tool_testsuite/config/migration_consumer.properties --- .../config/migration_consumer.properties | 4 ++-- 1 files changed, 2 insertions(+), 2 deletions(-) diff --git a/system_test/migration_tool_testsuite/config/migration_consumer.properties b/system_test/migration_tool_testsuite/config/migration_consumer.properties index 8458215..184f1de 100644 --- a/system_test/migration_tool_testsuite/config/migration_consumer.properties +++ b/system_test/migration_tool_testsuite/config/migration_consumer.properties @@ -20,10 +20,10 @@ zk.connect=127.0.0.1:2188 # timeout in ms for connecting to zookeeper -zk.connection.timeout.ms=1000000 +zk.connectiontimeout.ms=1000000 #consumer group id -group.id=test-consumer-group +groupid=test-consumer-group #consumer timeout #consumer.timeout.ms=5000 -- 1.7.1