What I've learned this week?: This week 2/2021

Last year started my adventure with Big Data and Hadoop ecosystem. I created my first simple Java application for Spark. It worked but I felt some lack. Where are tests for that? I looked around and I couldn't find any good example of module testing. I was looking for test example for hexagon application, this time in Spark, so: mocks on input and output and call facade main method to check behaviour.

Finally I found a solution. Like standard business application I extracted code which is a point of contact with external world and mocked it. In my case I was testing a simple application which consumes message from Kafka, transforms it and stores a result in one of two tables.

Bellow I presented a simple class dependence which correspond to my application.

The KafkaStream and HadoopRepository are input and output of my application. On the top of them there are services and facade where are implemented transformation and DAG model creation.

To increase difficulty of this task I created my first test in scala.

class TransformationSpec extends AnyFlatSpec 
                            with BeforeAndAfter 
                            with BeforeAndAfterEach
                            with should.Matchers 
                            with GivenWhenThen {

  private val master = "local[3]"
  private val appName = "test-app"
  private val batchDuration = Seconds(1)

  private val kafkaStream = mock[KafkaStream]
  private val rowRecordRepositoryMock = mock[HadoopRepository[RowRecord]]
  private val errorRecordRepositoryMock = mock[HadoopRepository[ErrorRecord]]
  private val serviceA: ServiceA = new ServiceA(rowRecordRepositoryMock, errorRecordRepositoryMock)
  private val serviceB: ServiceB = new ServiceB(kafkaStream)
  private val underTest: TransferDataService = new TransferDataService(serviceA, serviceB)

  private var sc: SparkContext = _
  private var ssc: StreamingContext = _
  private val lines = mutable.Queue[RDD[ConsumerRecord[String, ReceivedMsg]]]()

  before {
    Mockito.reset(kafkaStream, rowRecordRepositoryMock, errorRecordRepositoryMock)
    val conf = new SparkConf()
      .setMaster(master)
      .setAppName(appName)
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    ssc = new StreamingContext(conf, batchDuration)
    sc = ssc.sparkContext
    mockKafkaStream
  }

  after {
    if (ssc != null) {
      ssc.stop()
    }
  }
...

In example I am building application (lines 11-18) - injecting dependences. Then I define class variables for Spark context and Spark Streaming context and finally a queue to simulate message receiving.

In "before" method for each test state of all mocks are reset, Spark configuration, context and mock KafkaStream with specific behaviour are created. Below is shown how messages are added to queue and how KafkaStream is mocked using MockitoSugar library.

  private def receivedMessage(msg: ReceivedMsg) = {
    lines += sc.makeRDD(Seq(new ConsumerRecord("test", 0, 0, "", msg)))
  }

  private def mockKafkaStream = {
    MockitoSugar.doAnswer(() => {
      new JavaInputDStream[ConsumerRecord[String, ReceivedMsg]](
        ssc.queueStream(lines, true)
      )
    }).when(kafkaStream).createDirectStream(any());
  }

Now it is all ready to write first test case. I chose FlatSpec style. In "given" section I refilled message queue. In "when" section I called facade to initiate DAG model. I started Spark Streaming context and I run process.

In "then" section I asserted mock and captors. Application uses many threads so I couldn't call assertions just like that. To solve this problem scalatest provides method "eventually" to check conditions with some interval until match condition or meet timeout.

"Deserialized message" should "be stored in target table" in {

  Given("deserialized message")
  receivedMessage(new ReceivedMsg())

  When("Processing")
  underTest.process(new JavaStreamingContext(ssc));
  ssc.start()

  Then("message is stored by n records")
  eventually(timeout(2 second)) {
    val captor: ArgumentCaptor[JavaRDD[RowRecord]] = ArgumentCaptor.forClass(classOf[JavaRDD[RowRecord]])
    verify(rowRecordRepositoryMock)
      .save(captor.capture())

    val records = captor.getValue.collect()
    records should have size (4)

    verify(errorRecordRepositoryMock, never())
      .save(any())

    verify(kafkaStream)
      .commit(any(), any());
  }
}