Test sink behaviour

Posted by David Magalhães on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Test-sink-behaviour-tp32793.html

Hi, I've created a CustomSink that writes parquet file to S3. Inside the `invoke` method I have a loop to check if S3 is down, and if it is it will wait exponentially until it is online again.

Now I want to write a test for this, and I can execute everything and see that the Sink is doing what is suppose to do, but I can't have a way to validate that is doing that programmatically (in a integration test).

One of the possibilities I was thinking was check the LazyLogger errors, to verify that something was printed, but I can't mock Logger, since it is final. Since I expose the number of errors as a counter, I was trying to find a way to access it directly with Scala, but the only way I could find was via Rest API, and that is kind of a hack.

Exemple:

- Get the Rest API port with flinkCluster.getClusterClient.getFlinkConfiguration.getInteger("rest.port", 0)
- Get the jobId via http://localhost:61869/jobs/
- Get the verticeId via http://localhost:61869/jobs/c3879cca4ba23ad734b2810ba0d73873

- Get the metric via http://localhost:61869/jobs/c3879cca4ba23ad734b2810ba0d73873/vertices/0a448493b4782967b150582570326227/metrics/?get=0.Sink__Unnamed.errors_sink

Should be available a better way to get the metric or test this ?

Thanks