Skip to content

Commit

Permalink
SAMZA-2486: Update TestFileReaderSystemAdmin to use test resource fil…
Browse files Browse the repository at this point in the history
…es (apache#1315)

Symptom:
samza-core tests are randomly failing.

Cause:
TestFileReaderSystemAdmin and TestFileReaderSystemConsumer are creating and deleting the same set of files, when runnig in parallel, FileNotFoundException could happen depending on the timing.

Changes:
1. Update TestFileReaderSystemAdmin to use test resource files.
2. Update assert conditions as we removed tailing spaces and newlines in the files.
3. Keep TestFileReaderSystemConsumer as it is because it is also appending data to test files, thus cannot share to use the same test resource file.

Tests:
./gradlew test

API Changes:
None

Upgrade Instructions:
None

Usage Instructions:
None

Co-authored-by: Ke Wu <[email protected]>
  • Loading branch information
kw2542 and kw2542 authored Mar 16, 2020
1 parent b06cb3f commit 9bd3d61
Show file tree
Hide file tree
Showing 7 changed files with 38 additions and 44 deletions.
1 change: 1 addition & 0 deletions build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,7 @@ rat {
'README.md',
'RELEASE.md',
'samza-core/src/test/resources/classloader/samza-framework-api-classes.txt',
'samza-core/src/test/resources/*.txt',
'samza-test/src/main/resources/**',
'samza-hdfs/src/main/resources/**',
'samza-hdfs/src/test/resources/**',
Expand Down
Empty file.
4 changes: 4 additions & 0 deletions samza-core/src/test/resources/moreEnter.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
first line
second line
third line
other lines
1 change: 1 addition & 0 deletions samza-core/src/test/resources/noEnter.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
first line
2 changes: 2 additions & 0 deletions samza-core/src/test/resources/oneEnter.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
first line
second line
3 changes: 3 additions & 0 deletions samza-core/src/test/resources/twoEnter.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
first line
second line
other lines
Original file line number Diff line number Diff line change
Expand Up @@ -19,60 +19,38 @@

package org.apache.samza.system.filereader

import java.io.PrintWriter
import java.io.File

import org.apache.samza.system.SystemStreamPartition
import org.apache.samza.Partition
import org.apache.samza.system.SystemStreamMetadata.SystemStreamPartitionMetadata
import org.junit.Assert._
import org.junit.Test
import org.junit.Before
import org.junit.After
import org.scalatest.junit.AssertionsForJUnit

import scala.collection.mutable.HashMap
import scala.collection.JavaConverters._
import scala.collection.mutable

class TestFileReaderSystemAdmin extends AssertionsForJUnit {

val files = List("empty.txt", "noEnter.txt", "oneEnter.txt", "twoEnter.txt", "moreEnter.txt")

@Before
def createFiles {
files.foreach(file => {
val writer = new PrintWriter(new File(file))
file match {
case "empty.txt" =>
case "noEnter.txt" => writer.write("first line")
case "oneEnter.txt" => writer.write("first line \nsecond line")
case "twoEnter.txt" => writer.write("first line \nsecond line \nother lines")
case "moreEnter.txt" => writer.write("first line \nsecond line \nthird line \nother lines \n")
}
writer.close
})
}

@After
def deleteFiles {
files.foreach(file => (new File(file)).delete)
}
val files = List(
getClass.getResource("/empty.txt").getPath,
getClass.getResource("/noEnter.txt").getPath,
getClass.getResource("/oneEnter.txt").getPath,
getClass.getResource("/twoEnter.txt").getPath,
getClass.getResource("/moreEnter.txt").getPath)

@Test
def testGetOffsetsAfter {
val fileReaderSystemAdmin = new FileReaderSystemAdmin
val ssp1 = new SystemStreamPartition("file-reader", files(0), new Partition(0))
val ssp2 = new SystemStreamPartition("file-reader", files(1), new Partition(0))
val ssp3 = new SystemStreamPartition("file-reader", files(2), new Partition(0))
val ssp4 = new SystemStreamPartition("file-reader", files(3), new Partition(0))
val ssp5 = new SystemStreamPartition("file-reader", files(4), new Partition(0))

val offsets: java.util.Map[SystemStreamPartition, String] =
HashMap(ssp3 -> "0", ssp4 -> "12", ssp5 -> "25").asJava
mutable.HashMap(ssp3 -> "0", ssp4 -> "12", ssp5 -> "25").asJava
val afterOffsets = fileReaderSystemAdmin.getOffsetsAfter(offsets)
assertEquals("12", afterOffsets.get(ssp3))
assertEquals("25", afterOffsets.get(ssp4))
assertEquals("37", afterOffsets.get(ssp5))
assertEquals("11", afterOffsets.get(ssp3))
assertEquals("23", afterOffsets.get(ssp4))
assertEquals("34", afterOffsets.get(ssp5))
}

@Test
Expand All @@ -81,20 +59,25 @@ class TestFileReaderSystemAdmin extends AssertionsForJUnit {
val allMetadata = fileReaderSystemAdmin.getSystemStreamMetadata(setAsJavaSetConverter(files.toSet).asJava)
val expectedEmpty = new SystemStreamPartitionMetadata(null, null, "0")
val expectedNoEntry = new SystemStreamPartitionMetadata("0", "0", "0")
val expectedOneEntry = new SystemStreamPartitionMetadata("0", "0", "12")
val expectedTwoEntry = new SystemStreamPartitionMetadata("0", "12", "25")
val expectedMoreEntry = new SystemStreamPartitionMetadata("0", "37", "50")
val expectedOneEntry = new SystemStreamPartitionMetadata("0", "0", "11")
val expectedTwoEntry = new SystemStreamPartitionMetadata("0", "11", "23")
val expectedMoreEntry = new SystemStreamPartitionMetadata("0", "34", "46")

allMetadata.asScala.foreach { entry =>
{
val result = (entry._2).getSystemStreamPartitionMetadata().get(new Partition(0))
entry._1 match {
case "empty.txt" => assertEquals(expectedEmpty, result)
case "noEnter.txt" => assertEquals(expectedNoEntry, result)
case "oneEnter.txt" => assertEquals(expectedOneEntry, result)
case "twoEnter.txt" => assertEquals(expectedTwoEntry, result)
case "moreEnter.txt" => assertEquals(expectedMoreEntry, result)
}
val result = entry._2.getSystemStreamPartitionMetadata.get(new Partition(0))
if (entry._1.endsWith("empty.txt")) {
assertEquals(expectedEmpty, result)
} else if (entry._1.endsWith("noEnter.txt")) {
assertEquals(expectedNoEntry, result)
} else if (entry._1.endsWith("oneEnter.txt")) {
assertEquals(expectedOneEntry, result)
} else if (entry._1.endsWith("twoEnter.txt")) {
assertEquals(expectedTwoEntry, result)
} else if (entry._1.endsWith("moreEnter.txt")) {
assertEquals(expectedMoreEntry, result)
} else
fail()
}
}
}
Expand Down

0 comments on commit 9bd3d61

Please sign in to comment.