Improved Flink conformance test setup

sansa-inference-flink/src/test/scala/net/sansa_stack/inference/flink/conformance/SharedOWLHorstReasonerContext.scala

@@ -13,20 +13,13 @@ import org.scalatest.{BeforeAndAfterAll, Suite}
   * @author Lorenz Buehmann
   */
 @RunWith(classOf[Parameterized])
-trait SharedOWLHorstReasonerContext extends BeforeAndAfterAll with ReasonerContextProvider{
+trait SharedOWLHorstReasonerContext
+  extends SharedReasonerContext[ForwardRuleReasonerOWLHorst] {
   self: Suite =>
 
-  @transient private var _reasoner: ForwardRuleReasonerOWLHorst = _
-
-  val reasoner: ForwardRuleReasoner = _reasoner
-
-  @transient private var _env: ExecutionEnvironment = _
-  def env: ExecutionEnvironment = _env
-
 
   override def beforeAll(): Unit = {
     super.beforeAll()
-    _env = ExecutionEnvironment.getExecutionEnvironment
     _reasoner = new ForwardRuleReasonerOWLHorst(env)
   }
 }

sansa-inference-flink/src/test/scala/net/sansa_stack/inference/flink/conformance/SharedRDFSReasonerContext.scala

@@ -3,6 +3,7 @@ package net.sansa_stack.inference.flink.conformance
 import net.sansa_stack.inference.flink.forwardchaining.{ForwardRuleReasoner, ForwardRuleReasonerRDFS}
 import net.sansa_stack.inference.rules.RDFSLevel
 import org.apache.flink.api.scala.ExecutionEnvironment
+import org.apache.flink.test.util.AbstractTestBase
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
 import org.scalatest.{BeforeAndAfterAll, Suite}
@@ -13,20 +14,12 @@ import org.scalatest.{BeforeAndAfterAll, Suite}
   * @author Lorenz Buehmann
   */
 @RunWith(classOf[Parameterized])
-trait SharedRDFSReasonerContext extends BeforeAndAfterAll with ReasonerContextProvider{
+trait SharedRDFSReasonerContext
+    extends SharedReasonerContext[ForwardRuleReasonerRDFS] {
   self: Suite =>
 
-  @transient private var _reasoner: ForwardRuleReasonerRDFS = _
-  def reasoner: ForwardRuleReasoner = _reasoner
-
-  @transient private var _env: ExecutionEnvironment = _
-  def env: ExecutionEnvironment = _env
-
-
   override def beforeAll(): Unit = {
     super.beforeAll()
-    _env = ExecutionEnvironment.getExecutionEnvironment
-    _env.getConfig.disableSysoutLogging()
     _reasoner = new ForwardRuleReasonerRDFS(env)
     _reasoner.level = RDFSLevel.SIMPLE
   }

sansa-inference-flink/src/test/scala/net/sansa_stack/inference/flink/conformance/SharedReasonerContext.scala

@@ -0,0 +1,48 @@
+package net.sansa_stack.inference.flink.conformance
+
+import net.sansa_stack.rdf.common.kryo.jena.JenaKryoSerializers.{NodeSerializer, TripleSerializer}
+import org.apache.flink.api.scala.ExecutionEnvironment
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration
+import org.apache.jena.graph.{Node, Triple}
+import org.scalatest.{BeforeAndAfterAll, Suite}
+
+import net.sansa_stack.inference.flink.forwardchaining.ForwardRuleReasoner
+
+/**
+  * A shared reasoner and Flink environment used for multiple test cases using the same resources.
+  *
+  * @author Lorenz Buehmann
+  */
+trait SharedReasonerContext[R <: ForwardRuleReasoner]
+  extends BeforeAndAfterAll
+    with ReasonerContextProvider {
+  self: Suite =>
+
+  @transient protected var _reasoner: R = _
+  def reasoner: R = _reasoner
+
+  @transient private var _env: ExecutionEnvironment = _
+  def env: ExecutionEnvironment = _env
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    _env = ExecutionEnvironment.getExecutionEnvironment
+    _env.setParallelism(4)
+    _env.getConfig.disableSysoutLogging()
+    _env.getConfig.addDefaultKryoSerializer(classOf[Triple], classOf[TripleSerializer])
+    _env.getConfig.addDefaultKryoSerializer(classOf[Node], classOf[NodeSerializer])
+  }
+
+  import org.apache.flink.test.util.MiniClusterWithClientResource
+  import org.junit.ClassRule
+
+  private val DEFAULT_PARALLELISM = 4
+
+  @ClassRule val miniClusterResource = new MiniClusterWithClientResource(
+    new MiniClusterResourceConfiguration.Builder()
+      .setNumberTaskManagers(1)
+      .setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
+      .build
+  )
+
+}