-
Notifications
You must be signed in to change notification settings - Fork 3
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
NMS-13282: Fix traffic volume tests #29
Conversation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I would like to discuss if this way writing tests is something that we want. My main concern is twofold here:
-
By re-implementing/duplicating the code into the test and re-doing the calculations there we are not testing the correctness of the calculations but only the correctness of the underlying system. This maybe is something that we want, but the original purpose of the tests are to check the calculations itself.
-
The test structure has moved from a system test to something more like a integration test by interacting with the pipeline using a different approach. The oddities that the flink cluster (consisting of more than one node) may introduce are not tested anymore.
In addition, this PR is huge. I have no idea on how to review these changes in any sensible way.
IMHO that kind of test makes a lot of sense. However, if you insist I can remove that part from
I do not understand what your point is here. If the argument
If you closely look at the changes then you find that all of these changes are straight-forward:
|
I removed the e2e test part from the PR. |
@@ -152,7 +152,7 @@ public void tearDown() throws IOException { | |||
} | |||
|
|||
@Test | |||
public void canPee() throws Exception { | |||
public void testRates() throws Exception { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
awww, changing the name? LOL
private org.apache.beam.sdk.Pipeline createPipeline(NephronOptions options) { | ||
var pipeline = org.apache.beam.sdk.Pipeline.create(options); | ||
registerCoders(pipeline); | ||
Map<String, Object> kafkaConsumerConfig = new HashMap<>(); | ||
kafkaConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, options.getGroupId()); | ||
// Auto-commit should be disabled when checkpointing is on: | ||
// the state in the checkpoints are used to derive the offsets instead | ||
kafkaConsumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, options.getAutoCommit()); | ||
|
||
// use a timestamp policy that finishes processing when the input is idle for some time | ||
TimestampPolicyFactory<byte[], FlowDocument> tpf = timestampPolicyFactory( | ||
org.joda.time.Duration.millis(options.getDefaultMaxInputDelayMs()), | ||
org.joda.time.Duration.standardSeconds(5) | ||
); | ||
|
||
var readFromKafka = new Pipeline.ReadFromKafka( | ||
options.getBootstrapServers(), | ||
options.getFlowSourceTopic(), | ||
kafkaConsumerConfig, | ||
tpf | ||
); | ||
|
||
pipeline.apply(readFromKafka) | ||
.apply(new Pipeline.CalculateFlowStatistics(options)) | ||
.apply(new Pipeline.WriteToElasticsearch(options)); | ||
|
||
return pipeline; | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
In contrast to the "real" pipeline used in production: is this only required to patch in the changed timestamp policy?
Would be nice to avoid this - even if it results in longer test run time. Do we need the pipeline to end to flush things out? If not, why not let the pipeline run forever and kill it as soon as we got the expected results from ES?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, the pipeline creation is slightly tweaked only in order to allow the use of the special timestamp policy.
I prefer not to kill the pipeline. When we wait for results to appear in ES we never can be sure that all flow summaries for a specific timestamp have been persisted. (Some late panes may not yet have been persisted.) Waiting a couple of seconds may remedy this but how long should we wait? Circle-CI runs show sometimes big outliers in their required time.
Code duplication could be reduced by refactoring the pipeline creation logic.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Would be nice to avoid the code duplication if the effort is not to high.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Beside the possible removal of code duplication, this is fine.
fd11084
to
32e62bd
Compare
32e62bd
to
21fe293
Compare
Includes two minor changes:
and one significant change.
The significant change is that the
RandomFlowIT
now compares the volumes of flow summaries that are output by Nephron with expected results that are calculated in-memory. The generation rate is no more considered. This test does not test the flow generation logic but tests if the summaries calculated by Nephron are correct under various conditions.This requires that Nephron's calculation is not interrupted but finishes by itself because the end of its input is reached. This is accomplished by a specific timestamp policy factory that advances the watermark to the end when the input is idle for some amount of time.
The functionality of the
Handler
class is extracted into two independent pieces, namely theFlowDocGen
class that generates flow documents based on flow reports and theFlowDocSender
class that sends flow documents to Kafka. This split was necessary to do the in-memory-calculation that is based on flow documents.