#1307 Integration test for Streaming Conformance #1366

AdrianOlosutean · 2020-05-29T16:23:48Z

AdrianOlosutean · 2020-06-01T09:17:21Z

So far I've addded a test case with timeout which can catch #1306 . What other use cases could be useful for testing HyperConformance?

yruslan

A very interesting solution.
Questions from my side are:

Were you able to reproduce the Catalyst issue using this approach? E.g. conforming a dataset hangs the computation.
How long it takes to run one such test?
I'm wondering how it works in general, we can discuss this tomorrow.

spark-jobs/src/main/scala/za/co/absa/enceladus/conformance/HyperConformance.scala

...test/scala/za/co/absa/enceladus/conformance/streaming/HyperConformanceIntegrationSuite.scala

.../src/test/scala/za/co/absa/enceladus/conformance/interpreter/fixtures/StreamingFixture.scala

AdrianOlosutean · 2020-06-01T12:20:54Z

Yes, I was able to reproduce it by enabling or disabling the workaround and running it on the test. As @kevinwallimann said before, this bug is happening even if you provide just a MemoryStream.
The class seems to take around 30 seconds

kevinwallimann · 2020-06-01T12:34:19Z

.../src/test/scala/za/co/absa/enceladus/conformance/interpreter/fixtures/StreamingFixture.scala

+      .start()
+
+    input.collect().foreach(e => memoryStream.addData(e))
+    sink.awaitTermination(interval)


awaitTermination(interval) is probably a little bit dangerous. If the query is terminated just before the microbatch has been committed, then the whole microbatch will be aborted and there will be zero rows.
Also, you have to assume that your data will be processed within interval ms. 10s is probably enough, but it's kind of a magic value.
In this case processAllAvailable would be a good fit, because then you can be sure that all the rows will have been read, no matter to what value interval has been set to. processAllAvailable terminates as long as the data source is also finite which is the case here

I agreee, but how can you use processAllAvailable on a test with the catalyst optimizer workaround disabled without having an unreasonable running time for testing?

Ok, now I see your need for a timeout. Either you give up on testing with multiple micro-batches and use Trigger.Once or you could use a margin of safety, say 3 * interval.
Also, if you really want to cover multiple micro-batches, you need to make sure that the foreach loop runs during multiple micro-batches.

I'm slightly confused. We need to test 2 things:

The correctness of the results of streaming conformance. All workarounds should be turned on so such tests can finish in a reasonable time.

We need to test that workaround works by adding a test that hangs if the workaround is turned off. There is no particular time limit on the test, but should be some reasonable time.

Tests that test (1) should not rely on timing if possible. If processAllAvailable helps with it - great..
Tests that test (2) might rely on timing, but not necessarily. Currently in batch, if the workaround stops working the corresponding test will just hang forever. Not ideal, but we expect workaround to work. If it stops having an effect, it will be noticed by the developer when the test hangs.
Alternatively, a big timeout can be used to interrupt the test if it takes too long. But the value of the timeout should be set so that test passes on all envs the test could run, including Jenkins VMs.

The reason I care about this so much is that timings introduce non-determinism. Such tests could work perfectly locally on Mac, but start failing on Windows or on Linux build box.

.../src/test/scala/za/co/absa/enceladus/conformance/interpreter/fixtures/StreamingFixture.scala

Zejnilovic

Looks good. I love when tests are added, obviously. 👍

Zejnilovic · 2020-06-03T08:14:43Z

...test/scala/za/co/absa/enceladus/conformance/streaming/HyperConformanceIntegrationSuite.scala

+    val frame: DataFrame = testHyperConformance(standardizedDf,
+      "result",
+      nestedStructsDS)
+    frame.show()


Don't use show in tests

I mostly thought that show would make following what happens at the runtime easier

I understand, but in an ideal world, you just look at the bottom where it says that everything passed. In case of an issue, every one of us can write this line by themselves and re-run

Zejnilovic · 2020-06-03T08:16:17Z

...test/scala/za/co/absa/enceladus/conformance/streaming/HyperConformanceIntegrationSuite.scala

+    assertResult(frame.count())(20)
+  }
+
+  test("Test without catalyst workaround") {


I am not sure what this tests?

The idea is that since the workaround is disabled, the test will timeout and no data will be added to the sink in a reasonable time. Probably makes sense to add more description

This is testing of the presence of a bug is Spark. Seems quite weird to me and I would remove the test.

It might make sense to keep a commented test that, once uncommented, will demonstrate that the Catalyst issue still exists and the test hangs.
It might be useful in case new version of Spark has this issue fixed. We could try it then to see if the fix covers our use case.

Zejnilovic · 2020-06-03T08:27:25Z

.../src/test/scala/za/co/absa/enceladus/conformance/interpreter/fixtures/StreamingFixture.scala

+import za.co.absa.enceladus.utils.testUtils.SparkTestBase
+
+trait StreamingFixture extends FunSuite with SparkTestBase with MockitoSugar {
+  implicit val menasBaseUrls: List[String] = List()


I know it is a matter of taste, but readability wise I like List.empty[String] the most.

Considering it's an implicit that should have a type declaration, I would suggest.

Suggested change

implicit val menasBaseUrls: List[String] = List()

implicit val menasBaseUrls: List[String] = List.empty

.../test/scala/za/co/absa/enceladus/conformance/interpreter/fixtures/NestedStructsFixture.scala

yruslan

Looks good overall. Just a couple of suggestions.

.../src/test/scala/za/co/absa/enceladus/conformance/interpreter/fixtures/StreamingFixture.scala

...test/scala/za/co/absa/enceladus/conformance/streaming/HyperConformanceIntegrationSuite.scala

…ing-conformance-integration-test

yruslan

Looks great!

…ing-conformance-integration-test

sonarqubecloud · 2020-06-15T15:49:34Z

Kudos, SonarCloud Quality Gate passed!

0 Bugs
0 Vulnerabilities (and 0 Security Hotspots to review)
1 Code Smell

No Coverage information
0.0% Duplication

AdrianOlosutean · 2020-06-16T11:12:17Z

Release notes: here

#1307 Tests catching catalyst optimizer issue

71e5c83

AdrianOlosutean requested review from yruslan, kevinwallimann, benedeki, dk1844 and Zejnilovic and removed request for yruslan May 30, 2020 09:42

yruslan reviewed Jun 1, 2020

View reviewed changes

kevinwallimann reviewed Jun 1, 2020

View reviewed changes

#1307 Testing conformed data

31163c5

AdrianOlosutean marked this pull request as ready for review June 2, 2020 09:34

AdrianOlosutean requested review from DzMakatun, HuvarVer and lokm01 as code owners June 2, 2020 09:34

#1307 Testing with more logging and time between microbatches

6a661d0

AdrianOlosutean changed the title ~~#1307 Tests catching catalyst optimizer issue~~ #1307 Integration test for Streaming Conformance Jun 2, 2020

yruslan reviewed Jun 3, 2020

View reviewed changes

.../src/test/scala/za/co/absa/enceladus/conformance/interpreter/fixtures/StreamingFixture.scala Outdated Show resolved Hide resolved

Zejnilovic reviewed Jun 3, 2020

View reviewed changes

yruslan reviewed Jun 3, 2020

View reviewed changes

Adrian-Olosutean and others added 4 commits June 8, 2020 17:31

#1307 conformed complete test for literal

587d38d

#1307 Added test for eventtime column

ace740e

Merge branch 'develop' into feature/1307-add-kafka-source-sink-stream…

1d79a42

…ing-conformance-integration-test

#1307 Fixed wrong conformed_eventtime.json and reading pattern

5a520c7

AdrianOlosutean requested a review from yruslan June 11, 2020 16:29

yruslan approved these changes Jun 11, 2020

View reviewed changes

Zejnilovic approved these changes Jun 12, 2020

View reviewed changes

Merge branch 'develop' into feature/1307-add-kafka-source-sink-stream…

fd24091

…ing-conformance-integration-test

kevinwallimann approved these changes Jun 12, 2020

View reviewed changes

Merge branch 'develop' into feature/1307-add-kafka-source-sink-stream…

de959db

…ing-conformance-integration-test

AdrianOlosutean merged commit 29ce06c into develop Jun 17, 2020

AdrianOlosutean deleted the feature/1307-add-kafka-source-sink-streaming-conformance-integration-test branch June 17, 2020 11:45

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

#1307 Integration test for Streaming Conformance #1366

#1307 Integration test for Streaming Conformance #1366

AdrianOlosutean commented May 29, 2020

AdrianOlosutean commented Jun 1, 2020

yruslan left a comment

AdrianOlosutean commented Jun 1, 2020

kevinwallimann Jun 1, 2020 •

edited

Loading

AdrianOlosutean Jun 1, 2020

kevinwallimann Jun 1, 2020

yruslan Jun 3, 2020

Zejnilovic left a comment

Zejnilovic Jun 3, 2020

AdrianOlosutean Jun 3, 2020

Zejnilovic Jun 3, 2020

Zejnilovic Jun 3, 2020

AdrianOlosutean Jun 3, 2020

Zejnilovic Jun 3, 2020

yruslan Jun 3, 2020

Zejnilovic Jun 3, 2020

benedeki Jun 3, 2020 •

edited

Loading

yruslan left a comment

yruslan left a comment

sonarqubecloud bot commented Jun 15, 2020

AdrianOlosutean commented Jun 16, 2020

	implicit val menasBaseUrls: List[String] = List()
	implicit val menasBaseUrls: List[String] = List.empty

#1307 Integration test for Streaming Conformance #1366

#1307 Integration test for Streaming Conformance #1366

Conversation

AdrianOlosutean commented May 29, 2020

AdrianOlosutean commented Jun 1, 2020

yruslan left a comment

Choose a reason for hiding this comment

AdrianOlosutean commented Jun 1, 2020

kevinwallimann Jun 1, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Zejnilovic left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benedeki Jun 3, 2020 • edited Loading

Choose a reason for hiding this comment

yruslan left a comment

Choose a reason for hiding this comment

yruslan left a comment

Choose a reason for hiding this comment

sonarqubecloud bot commented Jun 15, 2020

AdrianOlosutean commented Jun 16, 2020

kevinwallimann Jun 1, 2020 •

edited

Loading

benedeki Jun 3, 2020 •

edited

Loading