Repository: kafka Updated Branches: refs/heads/trunk 1bfaddae9 -> 01aeea7c7
http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/main/scala/kafka/server/KafkaConfig.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index ae810eb..d13c872 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -27,12 +27,12 @@ import kafka.message.{BrokerCompressionCodec, CompressionCodec, Message, Message import kafka.utils.CoreUtils import org.apache.kafka.clients.CommonClientConfigs import org.apache.kafka.common.config.SaslConfigs - import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, SslConfigs} import org.apache.kafka.common.metrics.MetricsReporter import org.apache.kafka.common.protocol.SecurityProtocol +import org.apache.kafka.common.record.TimestampType -import scala.collection.{Map, immutable, JavaConverters} +import scala.collection.{JavaConverters, Map, immutable} import JavaConverters._ object Defaults { @@ -94,7 +94,8 @@ object Defaults { val LogFlushSchedulerIntervalMs = Long.MaxValue val LogFlushOffsetCheckpointIntervalMs = 60000 val LogPreAllocateEnable = false - val MessageFormatVersion = ApiVersion.latestVersion.toString() + // lazy val as `InterBrokerProtocolVersion` is defined later + lazy val MessageFormatVersion = InterBrokerProtocolVersion val NumRecoveryThreadsPerDataDir = 1 val AutoCreateTopicsEnable = true val MinInSyncReplicas = 1 @@ -423,11 +424,12 @@ object KafkaConfig { val NumRecoveryThreadsPerDataDirDoc = "The number of threads per data directory to be used for log recovery at startup and flushing at shutdown" val AutoCreateTopicsEnableDoc = "Enable auto creation of topic on the server" val MinInSyncReplicasDoc = "define the minimum number of replicas in ISR needed to satisfy a produce request with acks=all (or -1)" - val MessageFormatVersionDoc = "Specify the message format version the broker will use to append messages to the logs. The value should be a valid ApiVersion." + - "Some Examples are: 0.8.2, 0.9.0.0, 0.10.0. Check ApiVersion for detail. When setting the message format version, " + - "user certifies that all the existing messages on disk is at or below that version. Otherwise consumers before 0.10.0.0 will break." - val MessageTimestampTypeDoc = "Define whether the timestamp in the message is message create time or log append time. The value should be either" + - " \"CreateTime\" or \"LogAppendTime\"" + val MessageFormatVersionDoc = "Specify the message format version the broker will use to append messages to the logs. The value should be a valid ApiVersion. " + + "Some examples are: 0.8.2, 0.9.0.0, 0.10.0, check ApiVersion for more details. By setting a particular message format version, the " + + "user is certifying that all the existing messages on disk are smaller or equal than the specified version. Setting this value incorrectly " + + "will cause consumers with older versions to break as they will receive messages with a format that they don't understand." + val MessageTimestampTypeDoc = "Define whether the timestamp in the message is message create time or log append time. The value should be either " + + "`CreateTime` or `LogAppendTime`" val MessageTimestampDifferenceMaxMsDoc = "The maximum difference allowed between the timestamp when a broker receives " + "a message and the timestamp specified in the message. If message.timestamp.type=CreateTime, a message will be rejected " + "if the difference in timestamp exceeds this threshold. This configuration is ignored if message.timestamp.type=LogAppendTime." @@ -798,8 +800,11 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean) extends Abstra val logRetentionTimeMillis = getLogRetentionTimeMillis val minInSyncReplicas = getInt(KafkaConfig.MinInSyncReplicasProp) val logPreAllocateEnable: java.lang.Boolean = getBoolean(KafkaConfig.LogPreAllocateProp) - val messageFormatVersion = getString(KafkaConfig.MessageFormatVersionProp) - val messageTimestampType = getString(KafkaConfig.MessageTimestampTypeProp) + // We keep the user-provided String as `ApiVersion.apply` can choose a slightly different version (eg if `0.10.0` + // is passed, `0.10.0-IV0` may be picked) + val messageFormatVersionString = getString(KafkaConfig.MessageFormatVersionProp) + val messageFormatVersion = ApiVersion(messageFormatVersionString) + val messageTimestampType = TimestampType.forName(getString(KafkaConfig.MessageTimestampTypeProp)) val messageTimestampDifferenceMaxMs = getLong(KafkaConfig.MessageTimestampDifferenceMaxMsProp) /** ********* Replication configuration ***********/ @@ -821,7 +826,10 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean) extends Abstra val leaderImbalanceCheckIntervalSeconds = getLong(KafkaConfig.LeaderImbalanceCheckIntervalSecondsProp) val uncleanLeaderElectionEnable: java.lang.Boolean = getBoolean(KafkaConfig.UncleanLeaderElectionEnableProp) val interBrokerSecurityProtocol = SecurityProtocol.forName(getString(KafkaConfig.InterBrokerSecurityProtocolProp)) - val interBrokerProtocolVersion = ApiVersion(getString(KafkaConfig.InterBrokerProtocolVersionProp)) + // We keep the user-provided String as `ApiVersion.apply` can choose a slightly different version (eg if `0.10.0` + // is passed, `0.10.0-IV0` may be picked) + val interBrokerProtocolVersionString = getString(KafkaConfig.InterBrokerProtocolVersionProp) + val interBrokerProtocolVersion = ApiVersion(interBrokerProtocolVersionString) /** ********* Controlled shutdown configuration ***********/ val controlledShutdownMaxRetries = getInt(KafkaConfig.ControlledShutdownMaxRetriesProp) @@ -978,7 +986,7 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean) extends Abstra s"${KafkaConfig.AdvertisedListenersProp} protocols must be equal to or a subset of ${KafkaConfig.ListenersProp} protocols. " + s"Found ${advertisedListeners.keySet}. The valid options based on currently configured protocols are ${listeners.keySet}" ) - require(interBrokerProtocolVersion.onOrAfter(ApiVersion(messageFormatVersion)), - s"message.format.version $messageFormatVersion cannot be used when inter.broker.protocol.version is set to $interBrokerProtocolVersion") + require(interBrokerProtocolVersion >= messageFormatVersion, + s"message.format.version $messageFormatVersionString cannot be used when inter.broker.protocol.version is set to $interBrokerProtocolVersionString") } } http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/main/scala/kafka/server/KafkaServer.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index e3e185f..2203df9 100755 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -75,8 +75,8 @@ object KafkaServer { logProps.put(LogConfig.CompressionTypeProp, kafkaConfig.compressionType) logProps.put(LogConfig.UncleanLeaderElectionEnableProp, kafkaConfig.uncleanLeaderElectionEnable) logProps.put(LogConfig.PreAllocateEnableProp, kafkaConfig.logPreAllocateEnable) - logProps.put(LogConfig.MessageFormatVersionProp, kafkaConfig.messageFormatVersion) - logProps.put(LogConfig.MessageTimestampTypeProp, kafkaConfig.messageTimestampType) + logProps.put(LogConfig.MessageFormatVersionProp, kafkaConfig.messageFormatVersion.version) + logProps.put(LogConfig.MessageTimestampTypeProp, kafkaConfig.messageTimestampType.name) logProps.put(LogConfig.MessageTimestampDifferenceMaxMsProp, kafkaConfig.messageTimestampDifferenceMaxMs) logProps } @@ -200,7 +200,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr kafkaController.startup() /* start kafka coordinator */ - consumerCoordinator = GroupCoordinator.create(config, zkUtils, replicaManager, kafkaMetricsTime) + consumerCoordinator = GroupCoordinator(config, zkUtils, replicaManager, kafkaMetricsTime) consumerCoordinator.startup() /* Get the authorizer and initialize it if one is specified.*/ @@ -515,7 +515,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr val shutdownSucceeded = // Before 0.9.0.0, `ControlledShutdownRequest` did not contain `client_id` and it's a mandatory field in // `RequestHeader`, which is used by `NetworkClient` - if (config.interBrokerProtocolVersion.onOrAfter(KAFKA_0_9_0)) + if (config.interBrokerProtocolVersion >= KAFKA_0_9_0) networkClientControlledShutdown(config.controlledShutdownMaxRetries.intValue) else blockingChannelControlledShutdown(config.controlledShutdownMaxRetries.intValue) @@ -591,7 +591,9 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr val defaultProps = KafkaServer.copyKafkaConfigToLog(config) val defaultLogConfig = LogConfig(defaultProps) - val configs = AdminUtils.fetchAllTopicConfigs(zkUtils).mapValues(LogConfig.fromProps(defaultProps, _)) + val configs = AdminUtils.fetchAllTopicConfigs(zkUtils).map { case (topic, configs) => + topic -> LogConfig.fromProps(defaultProps, configs) + } // read the log configurations from zookeeper val cleanerConfig = CleanerConfig(numThreads = config.logCleanerThreads, dedupeBufferSize = config.logCleanerDedupeBufferSize, http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala index 2fdb46c..de7269f 100644 --- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala +++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala @@ -56,8 +56,8 @@ class ReplicaFetcherThread(name: String, type PD = PartitionData private val fetchRequestVersion: Short = - if (brokerConfig.interBrokerProtocolVersion.onOrAfter(KAFKA_0_10_0_IV0)) 2 - else if (brokerConfig.interBrokerProtocolVersion.onOrAfter(KAFKA_0_9_0)) 1 + if (brokerConfig.interBrokerProtocolVersion >= KAFKA_0_10_0_IV0) 2 + else if (brokerConfig.interBrokerProtocolVersion >= KAFKA_0_9_0) 1 else 0 private val socketTimeout: Int = brokerConfig.replicaSocketTimeoutMs private val replicaId = brokerConfig.brokerId http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/main/scala/kafka/server/ReplicaManager.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 16b8c3a..e388d98 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -431,20 +431,14 @@ class ReplicaManager(val config: KafkaConfig, fatal("Halting due to unrecoverable I/O error while handling produce request: ", e) Runtime.getRuntime.halt(1) (topicPartition, null) - case utpe: UnknownTopicOrPartitionException => - (topicPartition, LogAppendResult(LogAppendInfo.UnknownLogAppendInfo, Some(utpe))) - case nle: NotLeaderForPartitionException => - (topicPartition, LogAppendResult(LogAppendInfo.UnknownLogAppendInfo, Some(nle))) - case mtle: RecordTooLargeException => - (topicPartition, LogAppendResult(LogAppendInfo.UnknownLogAppendInfo, Some(mtle))) - case mstle: RecordBatchTooLargeException => - (topicPartition, LogAppendResult(LogAppendInfo.UnknownLogAppendInfo, Some(mstle))) - case imse: CorruptRecordException => - (topicPartition, LogAppendResult(LogAppendInfo.UnknownLogAppendInfo, Some(imse))) - case ime : InvalidMessageException => - (topicPartition, LogAppendResult(LogAppendInfo.UnknownLogAppendInfo, Some(ime))) - case itse : InvalidTimestampException => - (topicPartition, LogAppendResult(LogAppendInfo.UnknownLogAppendInfo, Some(itse))) + case e@ (_: UnknownTopicOrPartitionException | + _: NotLeaderForPartitionException | + _: RecordTooLargeException | + _: RecordBatchTooLargeException | + _: CorruptRecordException | + _: InvalidMessageException | + _: InvalidTimestampException) => + (topicPartition, LogAppendResult(LogAppendInfo.UnknownLogAppendInfo, Some(e))) case t: Throwable => BrokerTopicStats.getBrokerTopicStats(topicPartition.topic).failedProduceRequestRate.mark() BrokerTopicStats.getBrokerAllTopicsStats.failedProduceRequestRate.mark() @@ -573,9 +567,10 @@ class ReplicaManager(val config: KafkaConfig, } } - def getMessageFormatVersion(topicAndPartition: TopicAndPartition): Option[Byte] = { - getReplica(topicAndPartition.topic, topicAndPartition.partition).flatMap(_.log.map(_.config.messageFormatVersion)) - } + def getMessageFormatVersion(topicAndPartition: TopicAndPartition): Option[Byte] = + getReplica(topicAndPartition.topic, topicAndPartition.partition).flatMap { replica => + replica.log.map(_.config.messageFormatVersion.messageFormatVersion) + } def maybeUpdateMetadataCache(correlationId: Int, updateMetadataRequest: UpdateMetadataRequest, metadataCache: MetadataCache) { replicaStateChangeLock synchronized { http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/main/scala/kafka/tools/SimpleConsumerShell.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/tools/SimpleConsumerShell.scala b/core/src/main/scala/kafka/tools/SimpleConsumerShell.scala index e20b061..dda9697 100755 --- a/core/src/main/scala/kafka/tools/SimpleConsumerShell.scala +++ b/core/src/main/scala/kafka/tools/SimpleConsumerShell.scala @@ -220,7 +220,7 @@ object SimpleConsumerShell extends Logging { System.out.println("next offset = " + offset) val message = messageAndOffset.message val key = if(message.hasKey) Utils.readBytes(message.key) else null - val value = if (message.isNull()) null else Utils.readBytes(message.payload) + val value = if (message.isNull) null else Utils.readBytes(message.payload) formatter.writeTo(key, value, message.timestamp, message.timestampType, System.out) numMessagesConsumed += 1 } catch { http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/integration/kafka/api/BaseConsumerTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/integration/kafka/api/BaseConsumerTest.scala b/core/src/test/scala/integration/kafka/api/BaseConsumerTest.scala index a66fe35..19a8882 100644 --- a/core/src/test/scala/integration/kafka/api/BaseConsumerTest.scala +++ b/core/src/test/scala/integration/kafka/api/BaseConsumerTest.scala @@ -296,19 +296,19 @@ abstract class BaseConsumerTest extends IntegrationTestHarness with Logging { for (i <- 0 until numRecords) { val record = records.get(i) val offset = startingOffset + i - assertEquals(tp.topic(), record.topic()) - assertEquals(tp.partition(), record.partition()) + assertEquals(tp.topic, record.topic) + assertEquals(tp.partition, record.partition) if (timestampType == TimestampType.CREATE_TIME) { - assertEquals(timestampType, record.timestampType()) + assertEquals(timestampType, record.timestampType) val timestamp = startingTimestamp + i - assertEquals(timestamp.toLong, record.timestamp()) + assertEquals(timestamp.toLong, record.timestamp) } else - assertTrue(s"Got unexpected timestamp ${record.timestamp()}. Timestamp should be between [$startingTimestamp, $now}]", - record.timestamp() >= startingTimestamp && record.timestamp() <= now) - assertEquals(offset.toLong, record.offset()) + assertTrue(s"Got unexpected timestamp ${record.timestamp}. Timestamp should be between [$startingTimestamp, $now}]", + record.timestamp >= startingTimestamp && record.timestamp <= now) + assertEquals(offset.toLong, record.offset) val keyAndValueIndex = startingKeyAndValueIndex + i - assertEquals(s"key $keyAndValueIndex", new String(record.key())) - assertEquals(s"value $keyAndValueIndex", new String(record.value())) + assertEquals(s"key $keyAndValueIndex", new String(record.key)) + assertEquals(s"value $keyAndValueIndex", new String(record.value)) } } http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala index f8ad633..9bdbf6d 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala @@ -335,7 +335,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { val producer = TestUtils.createNewProducer(brokerList, securityProtocol = securityProtocol, trustStoreFile = trustStoreFile, retries = 0, lingerMs = Long.MaxValue, props = Some(producerProps)) (0 until numRecords).foreach { i => - producer.send(new ProducerRecord(tp.topic(), tp.partition(), i.toLong, s"key $i".getBytes, s"value $i".getBytes)) + producer.send(new ProducerRecord(tp.topic, tp.partition, i.toLong, s"key $i".getBytes, s"value $i".getBytes)) } producer.close() } http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/unit/kafka/consumer/ConsumerIteratorTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/consumer/ConsumerIteratorTest.scala b/core/src/test/scala/unit/kafka/consumer/ConsumerIteratorTest.scala index b8ad15b..ac7ce51 100755 --- a/core/src/test/scala/unit/kafka/consumer/ConsumerIteratorTest.scala +++ b/core/src/test/scala/unit/kafka/consumer/ConsumerIteratorTest.scala @@ -20,9 +20,11 @@ package kafka.consumer import java.util.concurrent._ import java.util.concurrent.atomic._ + +import kafka.common.LongRef + import scala.collection._ import org.junit.Assert._ - import kafka.message._ import kafka.server._ import kafka.utils.TestUtils._ @@ -64,7 +66,7 @@ class ConsumerIteratorTest extends KafkaServerTestHarness { def testConsumerIteratorDeduplicationDeepIterator() { val messageStrings = (0 until 10).map(_.toString).toList val messages = messageStrings.map(s => new Message(s.getBytes)) - val messageSet = new ByteBufferMessageSet(DefaultCompressionCodec, new AtomicLong(0), messages:_*) + val messageSet = new ByteBufferMessageSet(DefaultCompressionCodec, new LongRef(0), messages:_*) topicInfos(0).enqueue(messageSet) assertEquals(1, queue.size) @@ -88,7 +90,7 @@ class ConsumerIteratorTest extends KafkaServerTestHarness { def testConsumerIteratorDecodingFailure() { val messageStrings = (0 until 10).map(_.toString).toList val messages = messageStrings.map(s => new Message(s.getBytes)) - val messageSet = new ByteBufferMessageSet(NoCompressionCodec, new AtomicLong(0), messages:_*) + val messageSet = new ByteBufferMessageSet(NoCompressionCodec, new LongRef(0), messages:_*) topicInfos(0).enqueue(messageSet) assertEquals(1, queue.size) http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala b/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala index 587abd5..90e2b95 100644 --- a/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/GroupCoordinatorResponseTest.scala @@ -89,7 +89,7 @@ class GroupCoordinatorResponseTest extends JUnitSuite { EasyMock.expect(zkUtils.getPartitionAssignmentForTopics(Seq(GroupCoordinator.GroupMetadataTopicName))).andReturn(ret) EasyMock.replay(zkUtils) - groupCoordinator = GroupCoordinator.create(KafkaConfig.fromProps(props), zkUtils, replicaManager, new SystemTime) + groupCoordinator = GroupCoordinator(KafkaConfig.fromProps(props), zkUtils, replicaManager, new SystemTime) groupCoordinator.startup() // add the partition into the owned partition list http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/unit/kafka/log/CleanerTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/log/CleanerTest.scala b/core/src/test/scala/unit/kafka/log/CleanerTest.scala index 69218ba..3773233 100755 --- a/core/src/test/scala/unit/kafka/log/CleanerTest.scala +++ b/core/src/test/scala/unit/kafka/log/CleanerTest.scala @@ -21,7 +21,6 @@ import java.io.File import java.nio._ import java.nio.file.Paths import java.util.Properties -import java.util.concurrent.atomic.AtomicLong import kafka.common._ import kafka.message._ @@ -261,7 +260,7 @@ class CleanerTest extends JUnitSuite { log.append(TestUtils.singleMessageSet(payload = "hello".getBytes, key = "hello".getBytes)) // forward offset and append message to next segment at offset Int.MaxValue - val messageSet = new ByteBufferMessageSet(NoCompressionCodec, new AtomicLong(Int.MaxValue-1), + val messageSet = new ByteBufferMessageSet(NoCompressionCodec, new LongRef(Int.MaxValue - 1), new Message("hello".getBytes, "hello".getBytes, Message.NoTimestamp, Message.MagicValue_V1)) log.append(messageSet, assignOffsets = false) log.append(TestUtils.singleMessageSet(payload = "hello".getBytes, key = "hello".getBytes)) @@ -284,7 +283,7 @@ class CleanerTest extends JUnitSuite { log.append(TestUtils.singleMessageSet(payload = "hello".getBytes, key = "hello".getBytes)) groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue) - assertEquals(log.numberOfSegments-1, groups.size) + assertEquals(log.numberOfSegments - 1, groups.size) for (group <- groups) assertTrue("Relative offset greater than Int.MaxValue", group.last.index.lastOffset - group.head.index.baseOffset <= Int.MaxValue) checkSegmentOrder(groups) @@ -313,7 +312,7 @@ class CleanerTest extends JUnitSuite { assertEquals("Should have the expected number of messages in the map.", end-start, map.size) for(i <- start until end) assertEquals("Should find all the keys", i.toLong, map.get(key(i))) - assertEquals("Should not find a value too small", -1L, map.get(key(start-1))) + assertEquals("Should not find a value too small", -1L, map.get(key(start - 1))) assertEquals("Should not find a value too large", -1L, map.get(key(end))) } val segments = log.logSegments.toSeq @@ -455,11 +454,11 @@ class CleanerTest extends JUnitSuite { magicValue = Message.MagicValue_V1)) def unkeyedMessage(value: Int) = - new ByteBufferMessageSet(new Message(bytes=value.toString.getBytes)) + new ByteBufferMessageSet(new Message(bytes = value.toString.getBytes)) def deleteMessage(key: Int) = - new ByteBufferMessageSet(new Message(key=key.toString.getBytes, - bytes=null, + new ByteBufferMessageSet(new Message(key = key.toString.getBytes, + bytes = null, timestamp = Message.NoTimestamp, magicValue = Message.MagicValue_V1)) http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/unit/kafka/log/FileMessageSetTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/log/FileMessageSetTest.scala b/core/src/test/scala/unit/kafka/log/FileMessageSetTest.scala index 0179166..a3e5b2d 100644 --- a/core/src/test/scala/unit/kafka/log/FileMessageSetTest.scala +++ b/core/src/test/scala/unit/kafka/log/FileMessageSetTest.scala @@ -20,6 +20,8 @@ package kafka.log import java.io._ import java.nio._ import java.util.concurrent.atomic._ + +import kafka.common.LongRef import org.junit.Assert._ import kafka.utils.TestUtils._ import kafka.message._ @@ -103,7 +105,7 @@ class FileMessageSetTest extends BaseMessageSetTestCases { def testSearch() { // append a new message with a high offset val lastMessage = new Message("test".getBytes) - messageSet.append(new ByteBufferMessageSet(NoCompressionCodec, new AtomicLong(50), lastMessage)) + messageSet.append(new ByteBufferMessageSet(NoCompressionCodec, new LongRef(50), lastMessage)) var position = 0 assertEquals("Should be able to find the first message by its offset", OffsetPosition(0L, position), http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/unit/kafka/log/LogConfigTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala index 1be9e65..71e40da 100644 --- a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala @@ -53,19 +53,17 @@ class LogConfigTest { @Test def testFromPropsInvalid() { - LogConfig.configNames().foreach((name) => { - name match { - case LogConfig.UncleanLeaderElectionEnableProp => return - case LogConfig.RetentionBytesProp => assertPropertyInvalid(name, "not_a_number") - case LogConfig.RetentionMsProp => assertPropertyInvalid(name, "not_a_number" ) - case LogConfig.CleanupPolicyProp => assertPropertyInvalid(name, "true", "foobar"); - case LogConfig.MinCleanableDirtyRatioProp => assertPropertyInvalid(name, "not_a_number", "-0.1", "1.2") - case LogConfig.MinInSyncReplicasProp => assertPropertyInvalid(name, "not_a_number", "0", "-1") - case LogConfig.MessageFormatVersionProp => assertPropertyInvalid(name, "") - case positiveIntProperty => assertPropertyInvalid(name, "not_a_number", "-1") - } + LogConfig.configNames.foreach(name => name match { + case LogConfig.UncleanLeaderElectionEnableProp => assertPropertyInvalid(name, "not a boolean") + case LogConfig.RetentionBytesProp => assertPropertyInvalid(name, "not_a_number") + case LogConfig.RetentionMsProp => assertPropertyInvalid(name, "not_a_number" ) + case LogConfig.CleanupPolicyProp => assertPropertyInvalid(name, "true", "foobar"); + case LogConfig.MinCleanableDirtyRatioProp => assertPropertyInvalid(name, "not_a_number", "-0.1", "1.2") + case LogConfig.MinInSyncReplicasProp => assertPropertyInvalid(name, "not_a_number", "0", "-1") + case LogConfig.MessageFormatVersionProp => assertPropertyInvalid(name, "") + case positiveIntProperty => assertPropertyInvalid(name, "not_a_number", "-1") }) - } + } private def assertPropertyInvalid(name: String, values: AnyRef*) { values.foreach((value) => { http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/unit/kafka/log/LogManagerTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala index 91a4449..46bfbed 100755 --- a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala @@ -20,7 +20,6 @@ package kafka.log import java.io._ import java.util.Properties -import kafka.api.ApiVersion import kafka.common._ import kafka.server.OffsetCheckpoint import kafka.utils._ http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala b/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala index 7b80c27..edbfd99 100644 --- a/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala @@ -18,10 +18,13 @@ import org.junit.Assert._ import java.util.concurrent.atomic._ -import org.junit.{Test, After} + +import kafka.common.LongRef +import org.junit.{After, Test} import kafka.utils.TestUtils import kafka.message._ import kafka.utils.SystemTime + import scala.collection._ class LogSegmentTest { @@ -43,7 +46,7 @@ class LogSegmentTest { /* create a ByteBufferMessageSet for the given messages starting from the given offset */ def messages(offset: Long, messages: String*): ByteBufferMessageSet = { new ByteBufferMessageSet(compressionCodec = NoCompressionCodec, - offsetCounter = new AtomicLong(offset), + offsetCounter = new LongRef(offset), messages = messages.map(s => new Message(s.getBytes)):_*) } @@ -275,4 +278,4 @@ class LogSegmentTest { assertEquals(oldSize, size) assertEquals(size, fileSize) } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/unit/kafka/log/LogTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/log/LogTest.scala b/core/src/test/scala/unit/kafka/log/LogTest.scala index 426b5e8..5446eec 100755 --- a/core/src/test/scala/unit/kafka/log/LogTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogTest.scala @@ -20,8 +20,10 @@ package kafka.log import java.io._ import java.util.Properties import java.util.concurrent.atomic._ -import org.apache.kafka.common.errors.{OffsetOutOfRangeException, RecordBatchTooLargeException, RecordTooLargeException, CorruptRecordException} + +import org.apache.kafka.common.errors.{CorruptRecordException, OffsetOutOfRangeException, RecordBatchTooLargeException, RecordTooLargeException} import kafka.api.ApiVersion +import kafka.common.LongRef import org.junit.Assert._ import org.scalatest.junit.JUnitSuite import org.junit.{After, Before, Test} @@ -61,10 +63,10 @@ class LogTest extends JUnitSuite { */ @Test def testTimeBasedLogRoll() { - val set = TestUtils.singleMessageSet("test".getBytes()) + val set = TestUtils.singleMessageSet("test".getBytes) val logProps = new Properties() - logProps.put(LogConfig.SegmentMsProp, 1 * 60 * 60: java.lang.Long) + logProps.put(LogConfig.SegmentMsProp, (1 * 60 * 60L): java.lang.Long) // create a log val log = new Log(logDir, @@ -98,7 +100,7 @@ class LogTest extends JUnitSuite { */ @Test def testTimeBasedLogRollJitter() { - val set = TestUtils.singleMessageSet("test".getBytes()) + val set = TestUtils.singleMessageSet("test".getBytes) val maxJitter = 20 * 60L val logProps = new Properties() @@ -134,7 +136,7 @@ class LogTest extends JUnitSuite { val logProps = new Properties() logProps.put(LogConfig.SegmentBytesProp, segmentSize: java.lang.Integer) // We use need to use magic value 1 here because the test is message size sensitive. - logProps.put(LogConfig.MessageFormatVersionProp, ApiVersion.latestVersion.toString()) + logProps.put(LogConfig.MessageFormatVersionProp, ApiVersion.latestVersion.toString) // create a log val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time) assertEquals("There should be exactly 1 segment.", 1, log.numberOfSegments) @@ -164,7 +166,7 @@ class LogTest extends JUnitSuite { val logProps = new Properties() logProps.put(LogConfig.SegmentBytesProp, 71: java.lang.Integer) // We use need to use magic value 1 here because the test is message size sensitive. - logProps.put(LogConfig.MessageFormatVersionProp, ApiVersion.latestVersion.toString()) + logProps.put(LogConfig.MessageFormatVersionProp, ApiVersion.latestVersion.toString) val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time) val messages = (0 until 100 by 2).map(id => new Message(id.toString.getBytes)).toArray @@ -192,7 +194,7 @@ class LogTest extends JUnitSuite { // now test the case that we give the offsets and use non-sequential offsets for(i <- 0 until messages.length) - log.append(new ByteBufferMessageSet(NoCompressionCodec, new AtomicLong(messageIds(i)), messages = messages(i)), assignOffsets = false) + log.append(new ByteBufferMessageSet(NoCompressionCodec, new LongRef(messageIds(i)), messages = messages(i)), assignOffsets = false) for(i <- 50 until messageIds.max) { val idx = messageIds.indexWhere(_ >= i) val read = log.read(i, 100, None).messageSet.head @@ -350,7 +352,7 @@ class LogTest extends JUnitSuite { val logProps = new Properties() logProps.put(LogConfig.SegmentBytesProp, configSegmentSize: java.lang.Integer) // We use need to use magic value 1 here because the test is message size sensitive. - logProps.put(LogConfig.MessageFormatVersionProp, ApiVersion.latestVersion.toString()) + logProps.put(LogConfig.MessageFormatVersionProp, ApiVersion.latestVersion.toString) val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time) try { @@ -534,7 +536,7 @@ class LogTest extends JUnitSuite { */ @Test def testTruncateTo() { - val set = TestUtils.singleMessageSet("test".getBytes()) + val set = TestUtils.singleMessageSet("test".getBytes) val setSize = set.sizeInBytes val msgPerSeg = 10 val segmentSize = msgPerSeg * setSize // each segment will be 10 messages @@ -591,7 +593,7 @@ class LogTest extends JUnitSuite { */ @Test def testIndexResizingAtTruncation() { - val set = TestUtils.singleMessageSet("test".getBytes()) + val set = TestUtils.singleMessageSet("test".getBytes) val setSize = set.sizeInBytes val msgPerSeg = 10 val segmentSize = msgPerSeg * setSize // each segment will be 10 messages @@ -623,7 +625,7 @@ class LogTest extends JUnitSuite { val bogusIndex1 = Log.indexFilename(logDir, 0) val bogusIndex2 = Log.indexFilename(logDir, 5) - val set = TestUtils.singleMessageSet("test".getBytes()) + val set = TestUtils.singleMessageSet("test".getBytes) val logProps = new Properties() logProps.put(LogConfig.SegmentBytesProp, set.sizeInBytes * 5: java.lang.Integer) logProps.put(LogConfig.SegmentIndexBytesProp, 1000: java.lang.Integer) @@ -649,7 +651,7 @@ class LogTest extends JUnitSuite { */ @Test def testReopenThenTruncate() { - val set = TestUtils.singleMessageSet("test".getBytes()) + val set = TestUtils.singleMessageSet("test".getBytes) val logProps = new Properties() logProps.put(LogConfig.SegmentBytesProp, set.sizeInBytes * 5: java.lang.Integer) logProps.put(LogConfig.SegmentIndexBytesProp, 1000: java.lang.Integer) @@ -682,7 +684,7 @@ class LogTest extends JUnitSuite { */ @Test def testAsyncDelete() { - val set = TestUtils.singleMessageSet("test".getBytes()) + val set = TestUtils.singleMessageSet("test".getBytes) val asyncDeleteMs = 1000 val logProps = new Properties() logProps.put(LogConfig.SegmentBytesProp, set.sizeInBytes * 5: java.lang.Integer) @@ -724,7 +726,7 @@ class LogTest extends JUnitSuite { */ @Test def testOpenDeletesObsoleteFiles() { - val set = TestUtils.singleMessageSet("test".getBytes()) + val set = TestUtils.singleMessageSet("test".getBytes) val logProps = new Properties() logProps.put(LogConfig.SegmentBytesProp, set.sizeInBytes * 5: java.lang.Integer) logProps.put(LogConfig.SegmentIndexBytesProp, 1000: java.lang.Integer) @@ -771,7 +773,7 @@ class LogTest extends JUnitSuite { logProps.put(LogConfig.IndexIntervalBytesProp, 1: java.lang.Integer) logProps.put(LogConfig.MaxMessageBytesProp, 64*1024: java.lang.Integer) val config = LogConfig(logProps) - val set = TestUtils.singleMessageSet("test".getBytes()) + val set = TestUtils.singleMessageSet("test".getBytes) val recoveryPoint = 50L for(iteration <- 0 until 50) { // create a log and write some messages to it @@ -807,7 +809,7 @@ class LogTest extends JUnitSuite { logProps.put(LogConfig.MaxMessageBytesProp, 64*1024: java.lang.Integer) logProps.put(LogConfig.IndexIntervalBytesProp, 1: java.lang.Integer) val config = LogConfig(logProps) - val set = TestUtils.singleMessageSet("test".getBytes()) + val set = TestUtils.singleMessageSet("test".getBytes) val parentLogDir = logDir.getParentFile assertTrue("Data directory %s must exist", parentLogDir.isDirectory) val cleanShutdownFile = new File(parentLogDir, Log.CleanShutdownFile) http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/unit/kafka/message/ByteBufferMessageSetTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/message/ByteBufferMessageSetTest.scala b/core/src/test/scala/unit/kafka/message/ByteBufferMessageSetTest.scala index 77f5d65..758dad2 100644 --- a/core/src/test/scala/unit/kafka/message/ByteBufferMessageSetTest.scala +++ b/core/src/test/scala/unit/kafka/message/ByteBufferMessageSetTest.scala @@ -18,8 +18,8 @@ package kafka.message import java.nio._ -import java.util.concurrent.atomic.AtomicLong +import kafka.common.LongRef import kafka.utils.TestUtils import org.apache.kafka.common.errors.InvalidTimestampException import org.apache.kafka.common.record.TimestampType @@ -34,7 +34,7 @@ class ByteBufferMessageSetTest extends BaseMessageSetTestCases { @Test def testValidBytes() { { - val messages = new ByteBufferMessageSet(NoCompressionCodec, new Message("hello".getBytes()), new Message("there".getBytes())) + val messages = new ByteBufferMessageSet(NoCompressionCodec, new Message("hello".getBytes), new Message("there".getBytes)) val buffer = ByteBuffer.allocate(messages.sizeInBytes + 2) buffer.put(messages.buffer) buffer.putShort(4) @@ -51,25 +51,23 @@ class ByteBufferMessageSetTest extends BaseMessageSetTestCases { @Test def testValidBytesWithCompression() { - { - val messages = new ByteBufferMessageSet(DefaultCompressionCodec, new Message("hello".getBytes()), new Message("there".getBytes())) - val buffer = ByteBuffer.allocate(messages.sizeInBytes + 2) - buffer.put(messages.buffer) - buffer.putShort(4) - val messagesPlus = new ByteBufferMessageSet(buffer) - assertEquals("Adding invalid bytes shouldn't change byte count", messages.validBytes, messagesPlus.validBytes) - } + val messages = new ByteBufferMessageSet(DefaultCompressionCodec, new Message("hello".getBytes), new Message("there".getBytes)) + val buffer = ByteBuffer.allocate(messages.sizeInBytes + 2) + buffer.put(messages.buffer) + buffer.putShort(4) + val messagesPlus = new ByteBufferMessageSet(buffer) + assertEquals("Adding invalid bytes shouldn't change byte count", messages.validBytes, messagesPlus.validBytes) } @Test def testEquals() { - var messages = new ByteBufferMessageSet(DefaultCompressionCodec, new Message("hello".getBytes()), new Message("there".getBytes())) - var moreMessages = new ByteBufferMessageSet(DefaultCompressionCodec, new Message("hello".getBytes()), new Message("there".getBytes())) + var messages = new ByteBufferMessageSet(DefaultCompressionCodec, new Message("hello".getBytes), new Message("there".getBytes)) + var moreMessages = new ByteBufferMessageSet(DefaultCompressionCodec, new Message("hello".getBytes), new Message("there".getBytes)) assertTrue(messages.equals(moreMessages)) - messages = new ByteBufferMessageSet(NoCompressionCodec, new Message("hello".getBytes()), new Message("there".getBytes())) - moreMessages = new ByteBufferMessageSet(NoCompressionCodec, new Message("hello".getBytes()), new Message("there".getBytes())) + messages = new ByteBufferMessageSet(NoCompressionCodec, new Message("hello".getBytes), new Message("there".getBytes)) + moreMessages = new ByteBufferMessageSet(NoCompressionCodec, new Message("hello".getBytes), new Message("there".getBytes)) assertTrue(messages.equals(moreMessages)) } @@ -161,23 +159,26 @@ class ByteBufferMessageSetTest extends BaseMessageSetTestCases { val compressedMessagesWithoutRecompression = getMessages(magicValue = Message.MagicValue_V1, timestamp = -1L, codec = DefaultCompressionCodec) - val validatedMessages = messages.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(0), - sourceCodec = NoCompressionCodec, - targetCodec = NoCompressionCodec, - messageFormatVersion = 1, - messageTimestampType = TimestampType.LOG_APPEND_TIME, - messageTimestampDiffMaxMs = 1000L) - - val validatedCompressedMessages = - compressedMessagesWithRecompresion.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(0), + val (validatedMessages, _) = messages.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(0), + now = System.currentTimeMillis(), + sourceCodec = NoCompressionCodec, + targetCodec = NoCompressionCodec, + messageFormatVersion = 1, + messageTimestampType = TimestampType.LOG_APPEND_TIME, + messageTimestampDiffMaxMs = 1000L) + + val (validatedCompressedMessages, _) = + compressedMessagesWithRecompresion.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(0), + now = System.currentTimeMillis(), sourceCodec = DefaultCompressionCodec, targetCodec = DefaultCompressionCodec, messageFormatVersion = 1, messageTimestampType = TimestampType.LOG_APPEND_TIME, messageTimestampDiffMaxMs = 1000L) - val validatedCompressedMessagesWithoutRecompression = - compressedMessagesWithoutRecompression.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(0), + val (validatedCompressedMessagesWithoutRecompression, _) = + compressedMessagesWithoutRecompression.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(0), + now = System.currentTimeMillis(), sourceCodec = DefaultCompressionCodec, targetCodec = DefaultCompressionCodec, messageFormatVersion = 1, @@ -186,16 +187,15 @@ class ByteBufferMessageSetTest extends BaseMessageSetTestCases { val now = System.currentTimeMillis() assertEquals("message set size should not change", messages.size, validatedMessages.size) - validatedMessages.foreach({case messageAndOffset => validateLogAppendTime(messageAndOffset.message)}) + validatedMessages.foreach(messageAndOffset => validateLogAppendTime(messageAndOffset.message)) assertEquals("message set size should not change", compressedMessagesWithRecompresion.size, validatedCompressedMessages.size) - validatedCompressedMessages.foreach({case messageAndOffset => validateLogAppendTime(messageAndOffset.message)}) + validatedCompressedMessages.foreach(messageAndOffset => validateLogAppendTime(messageAndOffset.message)) assertTrue("MessageSet should still valid", validatedCompressedMessages.shallowIterator.next().message.isValid) assertEquals("message set size should not change", compressedMessagesWithoutRecompression.size, validatedCompressedMessagesWithoutRecompression.size) - validatedCompressedMessagesWithoutRecompression.foreach({case messageAndOffset => - validateLogAppendTime(messageAndOffset.message)}) + validatedCompressedMessagesWithoutRecompression.foreach(messageAndOffset => validateLogAppendTime(messageAndOffset.message)) assertTrue("MessageSet should still valid", validatedCompressedMessagesWithoutRecompression.shallowIterator.next().message.isValid) def validateLogAppendTime(message: Message) { @@ -212,15 +212,17 @@ class ByteBufferMessageSetTest extends BaseMessageSetTestCases { val messages = getMessages(magicValue = Message.MagicValue_V1, timestamp = now, codec = NoCompressionCodec) val compressedMessages = getMessages(magicValue = Message.MagicValue_V1, timestamp = now, codec = DefaultCompressionCodec) - val validatedMessages = messages.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(0), - sourceCodec = NoCompressionCodec, - targetCodec = NoCompressionCodec, - messageFormatVersion = 1, - messageTimestampType = TimestampType.CREATE_TIME, - messageTimestampDiffMaxMs = 1000L) - - val validatedCompressedMessages = - compressedMessages.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(0), + val (validatedMessages, _) = messages.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(0), + now = System.currentTimeMillis(), + sourceCodec = NoCompressionCodec, + targetCodec = NoCompressionCodec, + messageFormatVersion = 1, + messageTimestampType = TimestampType.CREATE_TIME, + messageTimestampDiffMaxMs = 1000L) + + val (validatedCompressedMessages, _) = + compressedMessages.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(0), + now = System.currentTimeMillis(), sourceCodec = DefaultCompressionCodec, targetCodec = DefaultCompressionCodec, messageFormatVersion = 1, @@ -246,7 +248,8 @@ class ByteBufferMessageSetTest extends BaseMessageSetTestCases { val compressedMessages = getMessages(magicValue = Message.MagicValue_V1, timestamp = now - 1001L, codec = DefaultCompressionCodec) try { - messages.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(0), + messages.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(0), + now = System.currentTimeMillis(), sourceCodec = NoCompressionCodec, targetCodec = NoCompressionCodec, messageFormatVersion = 1, @@ -258,7 +261,8 @@ class ByteBufferMessageSetTest extends BaseMessageSetTestCases { } try { - compressedMessages.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(0), + compressedMessages.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(0), + now = System.currentTimeMillis(), sourceCodec = DefaultCompressionCodec, targetCodec = DefaultCompressionCodec, messageFormatVersion = 1, @@ -277,21 +281,23 @@ class ByteBufferMessageSetTest extends BaseMessageSetTestCases { // check uncompressed offsets checkOffsets(messages, 0) val offset = 1234567 - checkOffsets(messages.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(offset), + checkOffsets(messages.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(offset), + now = System.currentTimeMillis(), sourceCodec = NoCompressionCodec, targetCodec = NoCompressionCodec, messageFormatVersion = 0, messageTimestampType = TimestampType.CREATE_TIME, - messageTimestampDiffMaxMs = 1000L), offset) + messageTimestampDiffMaxMs = 1000L)._1, offset) // check compressed messages checkOffsets(compressedMessages, 0) - checkOffsets(compressedMessages.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(offset), + checkOffsets(compressedMessages.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(offset), + now = System.currentTimeMillis(), sourceCodec = DefaultCompressionCodec, targetCodec = DefaultCompressionCodec, messageFormatVersion = 0, messageTimestampType = TimestampType.CREATE_TIME, - messageTimestampDiffMaxMs = 1000L), offset) + messageTimestampDiffMaxMs = 1000L)._1, offset) } @@ -304,20 +310,22 @@ class ByteBufferMessageSetTest extends BaseMessageSetTestCases { // check uncompressed offsets checkOffsets(messages, 0) val offset = 1234567 - val messageWithOffset = messages.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(offset), - sourceCodec = NoCompressionCodec, - targetCodec = NoCompressionCodec, - messageTimestampType = TimestampType.CREATE_TIME, - messageTimestampDiffMaxMs = 5000L) + val (messageWithOffset, _) = messages.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(offset), + now = System.currentTimeMillis(), + sourceCodec = NoCompressionCodec, + targetCodec = NoCompressionCodec, + messageTimestampType = TimestampType.CREATE_TIME, + messageTimestampDiffMaxMs = 5000L) checkOffsets(messageWithOffset, offset) // check compressed messages checkOffsets(compressedMessages, 0) - val compressedMessagesWithOffset = compressedMessages.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(offset), - sourceCodec = DefaultCompressionCodec, - targetCodec = DefaultCompressionCodec, - messageTimestampType = TimestampType.CREATE_TIME, - messageTimestampDiffMaxMs = 5000L) + val (compressedMessagesWithOffset, _) = compressedMessages.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(offset), + now = System.currentTimeMillis(), + sourceCodec = DefaultCompressionCodec, + targetCodec = DefaultCompressionCodec, + messageTimestampType = TimestampType.CREATE_TIME, + messageTimestampDiffMaxMs = 5000L) checkOffsets(compressedMessagesWithOffset, offset) } @@ -329,21 +337,23 @@ class ByteBufferMessageSetTest extends BaseMessageSetTestCases { // check uncompressed offsets checkOffsets(messagesV0, 0) val offset = 1234567 - checkOffsets(messagesV0.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(offset), + checkOffsets(messagesV0.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(offset), + now = System.currentTimeMillis(), sourceCodec = NoCompressionCodec, targetCodec = NoCompressionCodec, messageFormatVersion = 1, messageTimestampType = TimestampType.LOG_APPEND_TIME, - messageTimestampDiffMaxMs = 1000L), offset) + messageTimestampDiffMaxMs = 1000L)._1, offset) // check compressed messages checkOffsets(compressedMessagesV0, 0) - checkOffsets(compressedMessagesV0.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(offset), + checkOffsets(compressedMessagesV0.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(offset), + now = System.currentTimeMillis(), sourceCodec = DefaultCompressionCodec, targetCodec = DefaultCompressionCodec, messageFormatVersion = 1, messageTimestampType = TimestampType.LOG_APPEND_TIME, - messageTimestampDiffMaxMs = 1000L), offset) + messageTimestampDiffMaxMs = 1000L)._1, offset) // Check down conversion val now = System.currentTimeMillis() @@ -352,21 +362,23 @@ class ByteBufferMessageSetTest extends BaseMessageSetTestCases { // check uncompressed offsets checkOffsets(messagesV1, 0) - checkOffsets(messagesV1.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(offset), + checkOffsets(messagesV1.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(offset), + now = System.currentTimeMillis(), sourceCodec = NoCompressionCodec, targetCodec = NoCompressionCodec, messageFormatVersion = 0, messageTimestampType = TimestampType.CREATE_TIME, - messageTimestampDiffMaxMs = 5000L), offset) + messageTimestampDiffMaxMs = 5000L)._1, offset) // check compressed messages checkOffsets(compressedMessagesV1, 0) - checkOffsets(compressedMessagesV1.validateMessagesAndAssignOffsets(offsetCounter = new AtomicLong(offset), + checkOffsets(compressedMessagesV1.validateMessagesAndAssignOffsets(offsetCounter = new LongRef(offset), + now = System.currentTimeMillis(), sourceCodec = DefaultCompressionCodec, targetCodec = DefaultCompressionCodec, messageFormatVersion = 0, messageTimestampType = TimestampType.CREATE_TIME, - messageTimestampDiffMaxMs = 5000L), offset) + messageTimestampDiffMaxMs = 5000L)._1, offset) } /* check that offsets are assigned based on byte offset from the given base offset */ http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala index 7fe9ffc..aac50bd 100755 --- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala @@ -295,7 +295,7 @@ class KafkaConfigTest { assertEquals(KAFKA_0_8_2, conf3.interBrokerProtocolVersion) //check that latest is newer than 0.8.2 - assert(ApiVersion.latestVersion.onOrAfter(conf3.interBrokerProtocolVersion)) + assert(ApiVersion.latestVersion >= conf3.interBrokerProtocolVersion) } private def isValidKafkaConfig(props: Properties): Boolean = { http://git-wip-us.apache.org/repos/asf/kafka/blob/01aeea7c/docs/upgrade.html ---------------------------------------------------------------------- diff --git a/docs/upgrade.html b/docs/upgrade.html index f6d67eb..3370822 100644 --- a/docs/upgrade.html +++ b/docs/upgrade.html @@ -18,15 +18,15 @@ <h3><a id="upgrade" href="#upgrade">1.5 Upgrading From Previous Versions</a></h3> <h4><a id="upgrade_10" href="#upgrade_10">Upgrading from 0.8.x or 0.9.x to 0.10.0.0</a></h4> -0.10.0.0 has <a href="#upgrade_10_performance_impact">potential performance impact during upgrade</a> and -<a href="#upgrade_10_breaking">potential breaking changes</a> (please review before upgrading). Because new protocols +0.10.0.0 has <a href="#upgrade_10_breaking">potential breaking changes</a> (please review before upgrading) and +there may be a <a href="#upgrade_10_performance_impact">performance impact during the upgrade</a>. Because new protocols are introduced, it is important to upgrade your Kafka clusters before upgrading your clients. <p><b>For a rolling upgrade:</b></p> <ol> - <li> Update server.properties file on all brokers and add the following property: inter.broker.protocol.version=CURRENT_KAFKA_VERSION(e.g. 0.8.2, 0.9.0.0). - We recommend the users to set message.format.version=CURRENT_KAFKA_VERSION as well to avoid performance regression + <li> Update server.properties file on all brokers and add the following property: inter.broker.protocol.version=CURRENT_KAFKA_VERSION (e.g. 0.8.2 or 0.9.0.0). + We recommend that users set message.format.version=CURRENT_KAFKA_VERSION as well to avoid a performance regression during upgrade. See <a href="#upgrade_10_performance_impact">potential performance impact during upgrade</a> for the details. </li> <li> Upgrade the brokers. This can be done a broker at a time by simply bringing it down, updating the code, and restarting it. </li> @@ -38,35 +38,36 @@ are introduced, it is important to upgrade your Kafka clusters before upgrading <p><b>Note:</b> Bumping the protocol version and restarting can be done any time after the brokers were upgraded. It does not have to be immediately after. -<h5><a id="upgrade_10_performance_impact" href="#upgrade_10_performance_impact">potential performance impact in 0.10.0.0 during upgrade</a></h5> +<h5><a id="upgrade_10_performance_impact" href="#upgrade_10_performance_impact">Potential performance impact during upgrade to 0.10.0.0</a></h5> <p> - Message format in 0.10.0 now includes a new timestamp field and uses relative offsets for compressed messages. - The on disk message format can be configured through message.format.version in server.properties file. + The message format in 0.10.0 includes a new timestamp field and uses relative offsets for compressed messages. + The on disk message format can be configured through message.format.version in the server.properties file. The default on-disk message format is 0.10.0. If a consumer client is on a version before 0.10.0.0, it only understands - message format before 0.10.0. In this case, the broker is able to convert messages of the format in 0.10.0 to earlier format - before sending a response to the consumer on an older version. However, the broker can't use zero-copy transfer in this case. + message formats before 0.10.0. In this case, the broker is able to convert messages from the 0.10.0 format to an earlier format + before sending the response to the consumer on an older version. However, the broker can't use zero-copy transfer in this case. - To avoid such message conversion before consumers are upgraded to 0.10.0.0, one can set the message format to 0.9.0 when - upgrading the broker to 0.10.0.0. This way, the broker can still use zero-copy transfer to send the data to the old - consumers. Once most consumers are upgraded, one can change the message format to 0.10.0 on the broker. + To avoid such message conversion before consumers are upgraded to 0.10.0.0, one can set the message format to + e.g. 0.9.0 when upgrading the broker to 0.10.0.0. This way, the broker can still use zero-copy transfer to send the + data to the old consumers. Once most consumers are upgraded, one can change the message format to 0.10.0 on the broker. </p> <p> For clients that are upgraded to 0.10.0.0, there is no performance impact. </p> <p> - <b>Note:</b> By setting the message format version, one certifies all the existing messages are on or below that + <b>Note:</b> By setting the message format version, one certifies that all existing messages are on or below that message format version. Otherwise consumers before 0.10.0.0 might break. In particular, after the message format - is set to 0.10.0, one should not change it back to earlier format since it may break the consumer on versions before 0.10.0.0. + is set to 0.10.0, one should not change it back to an earlier format as it may break consumers on versions before 0.10.0.0. </p> <h5><a id="upgrade_10_breaking" href="#upgrade_10_breaking">potential breaking changes in 0.10.0.0</a></h5> <ul> - <li> Starting from Kafka 0.10.0.0, message format version in Kafka is represented as the Kafka version. For example, message format 0.9.0 refers to the highest message version supported by Kafka 0.9.0. </li> - <li> Message format 0.10.0 is added and used by default to include a timestamp field in the messages and use relative offsets for compressed messages. </li> - <li> ProduceRequest/Response v2 is added and used by default to support message format 0.10.0 </li> - <li> FetchRequest/Response v2 is added and used by default to support message format 0.10.0 </li> - <li> MessageFormatter interface changed from <code>void writeTo(byte[] key, byte[] value, PrintStream output)</code> to - <code>void writeTo(byte[] key, byte[] value, long timestamp, TimestampType timestampType, PrintStream output)</code> </li> + <li> Starting from Kafka 0.10.0.0, the message format version in Kafka is represented as the Kafka version. For example, message format 0.9.0 refers to the highest message version supported by Kafka 0.9.0. </li> + <li> Message format 0.10.0 has been introduced and it is used by default. It includes a timestamp field in the messages and relative offsets are used for compressed messages. </li> + <li> ProduceRequest/Response v2 has been introduced and it is used by default to support message format 0.10.0 </li> + <li> FetchRequest/Response v2 has been introduced and it is used by default to support message format 0.10.0 </li> + <li> MessageFormatter interface was changed from <code>void writeTo(byte[] key, byte[] value, PrintStream output)</code> to + <code>void writeTo(byte[] key, byte[] value, long timestamp, TimestampType timestampType, PrintStream output)</code> </li> + <li> MirrorMakerMessageHandler no longer exposes <em>handle(record: MessageAndMetadata[Array[Byte], Array[Byte]])</em> method as it was never called. </li> </ul> <h4><a id="upgrade_9" href="#upgrade_9">Upgrading from 0.8.0, 0.8.1.X or 0.8.2.X to 0.9.0.0</a></h4> @@ -107,9 +108,8 @@ are introduced, it is important to upgrade your Kafka clusters before upgrading <h5><a id="upgrade_901_notable" href="#upgrade_901_notable">Notable changes in 0.9.0.1</a></h5> <ul> - <li> The new broker id generation feature can be disable by setting broker.id.generation.enable to false. </li> + <li> The new broker id generation feature can be disabled by setting broker.id.generation.enable to false. </li> <li> Configuration parameter log.cleaner.enable is now true by default. This means topics with a cleanup.policy=compact will now be compacted by default, and 128 MB of heap will be allocated to the cleaner process via log.cleaner.dedupe.buffer.size. You may want to review log.cleaner.dedupe.buffer.size and the other log.cleaner configuration values based on your usage of compacted topics. </li> - <li> MirrorMakerMessageHandler no longer exposes <em>handle(record: MessageAndMetadata[Array[Byte], Array[Byte]])</em> method as it was never called. </li> <li> Default value of configuration parameter fetch.min.bytes for the new consumer is now 1 by default. </li> </ul>
