You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2016/03/15 20:15:19 UTC
kafka git commit: MINOR: Add test that verifies fix for KAFKA-3047
Repository: kafka
Updated Branches:
refs/heads/trunk 355076cd2 -> 90c623d10
MINOR: Add test that verifies fix for KAFKA-3047
Also clean-up `LogTest` a little.
Author: Ismael Juma <is...@juma.me.uk>
Reviewers: Guozhang Wang
Closes #1071 from ijuma/kafka-3047-explicit-offset-assignment-corrupt-log-test
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/90c623d1
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/90c623d1
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/90c623d1
Branch: refs/heads/trunk
Commit: 90c623d10f1c081e41f151eb28524f0fb57609c7
Parents: 355076c
Author: Ismael Juma <is...@juma.me.uk>
Authored: Tue Mar 15 12:15:15 2016 -0700
Committer: Guozhang Wang <wa...@gmail.com>
Committed: Tue Mar 15 12:15:15 2016 -0700
----------------------------------------------------------------------
.../src/test/scala/unit/kafka/log/LogTest.scala | 55 ++++++++++++--------
1 file changed, 34 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/90c623d1/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 5446eec..c2eb817 100755
--- a/core/src/test/scala/unit/kafka/log/LogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogTest.scala
@@ -321,7 +321,7 @@ class LogTest extends JUnitSuite {
for(i <- 0 until messagesToAppend)
log.append(TestUtils.singleMessageSet(i.toString.getBytes))
- var currOffset = log.logEndOffset
+ val currOffset = log.logEndOffset
assertEquals(currOffset, messagesToAppend)
// time goes by; the log file is deleted
@@ -765,6 +765,19 @@ class LogTest extends JUnitSuite {
assertTrue("Message payload should be null.", messageSet.head.message.isNull)
}
+ @Test(expected = classOf[IllegalArgumentException])
+ def testAppendWithOutOfOrderOffsetsThrowsException() {
+ val log = new Log(logDir,
+ LogConfig(),
+ recoveryPoint = 0L,
+ time.scheduler,
+ time)
+ val messages = (0 until 2).map(id => new Message(id.toString.getBytes)).toArray
+ messages.foreach(message => log.append(new ByteBufferMessageSet(message)))
+ val invalidMessage = new ByteBufferMessageSet(new Message(1.toString.getBytes))
+ log.append(invalidMessage, assignOffsets = false)
+ }
+
@Test
def testCorruptLog() {
// append some messages to create some segments
@@ -836,9 +849,9 @@ class LogTest extends JUnitSuite {
@Test
def testParseTopicPartitionName() {
- val topic: String = "test_topic"
- val partition:String = "143"
- val dir: File = new File(logDir + topicPartitionName(topic, partition))
+ val topic = "test_topic"
+ val partition = "143"
+ val dir = new File(logDir + topicPartitionName(topic, partition))
val topicAndPartition = Log.parseTopicPartitionName(dir)
assertEquals(topic, topicAndPartition.asTuple._1)
assertEquals(partition.toInt, topicAndPartition.asTuple._2)
@@ -847,8 +860,8 @@ class LogTest extends JUnitSuite {
@Test
def testParseTopicPartitionNameForEmptyName() {
try {
- val dir: File = new File("")
- val topicAndPartition = Log.parseTopicPartitionName(dir)
+ val dir = new File("")
+ Log.parseTopicPartitionName(dir)
fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath)
} catch {
case e: Exception => // its GOOD!
@@ -859,7 +872,7 @@ class LogTest extends JUnitSuite {
def testParseTopicPartitionNameForNull() {
try {
val dir: File = null
- val topicAndPartition = Log.parseTopicPartitionName(dir)
+ Log.parseTopicPartitionName(dir)
fail("KafkaException should have been thrown for dir: " + dir)
} catch {
case e: Exception => // its GOOD!
@@ -868,11 +881,11 @@ class LogTest extends JUnitSuite {
@Test
def testParseTopicPartitionNameForMissingSeparator() {
- val topic: String = "test_topic"
- val partition:String = "1999"
- val dir: File = new File(logDir + File.separator + topic + partition)
+ val topic = "test_topic"
+ val partition = "1999"
+ val dir = new File(logDir + File.separator + topic + partition)
try {
- val topicAndPartition = Log.parseTopicPartitionName(dir)
+ Log.parseTopicPartitionName(dir)
fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath)
} catch {
case e: Exception => // its GOOD!
@@ -881,11 +894,11 @@ class LogTest extends JUnitSuite {
@Test
def testParseTopicPartitionNameForMissingTopic() {
- val topic: String = ""
- val partition:String = "1999"
- val dir: File = new File(logDir + topicPartitionName(topic, partition))
+ val topic = ""
+ val partition = "1999"
+ val dir = new File(logDir + topicPartitionName(topic, partition))
try {
- val topicAndPartition = Log.parseTopicPartitionName(dir)
+ Log.parseTopicPartitionName(dir)
fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath)
} catch {
case e: Exception => // its GOOD!
@@ -894,18 +907,18 @@ class LogTest extends JUnitSuite {
@Test
def testParseTopicPartitionNameForMissingPartition() {
- val topic: String = "test_topic"
- val partition:String = ""
- val dir: File = new File(logDir + topicPartitionName(topic, partition))
+ val topic = "test_topic"
+ val partition = ""
+ val dir = new File(logDir + topicPartitionName(topic, partition))
try {
- val topicAndPartition = Log.parseTopicPartitionName(dir)
+ Log.parseTopicPartitionName(dir)
fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath)
} catch {
case e: Exception => // its GOOD!
}
}
- def topicPartitionName(topic: String, partition: String): String = {
+ def topicPartitionName(topic: String, partition: String): String =
File.separator + topic + "-" + partition
- }
+
}