title: How to use Groovy and Spock to test Apach Flink Job?
body: <p>I have a <code>Flink</code> job that reads data from <code>Kafka</code> into a table which is emitted into a <code>DataStream</code> on which I apply a filter function and then convert the data stream back to a table which writes data back to <code>Kafka</code>.</p>
  <p>I want to test the functionality of the filter function. I am writing unit tests in <code>Groovy</code> using <code>Spock</code> (framework). In my unit test I am calling the <code>Flink</code> job with the Table SQL string with details about the <code>Kafka</code> topic, however, I am confused on how to load the right <code>StreamExecution</code> and <code>TableEnvironment</code> because when I create a new object of my <code>Flink</code> class, those values are <code>null</code> and I don't have getters/setters to set everything up because that would make the code really messy.</p>
  <p>The following is my logic. My question is can I write <code>Apache Flink</code> APIs as seamlessly in <code>Groovy</code> or there are many layers/pitfalls and how can I better approach these tests:</p>
  <pre><code>class DataStreamTests extends Specification {
      @Autowired
      ApplicationConfiguration configuration;
      FlinkStreaming streaming = new FlinkStreaming();
      final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      EnvironmentSettings settings = EnvironmentSettings
              .newInstance()
              .inStreamingMode()
              .build();
      final StreamTableEnvironment tableEnvironment = TableEnvironment.create(settings);
      
      def resourcePath = &quot;cars/porche.txt&quot;
      StreamSpec streamSpec = ConfigurationParser.createStreamSpec(getClass().getResource(resourcePath).text)
      
      def &quot;create a new input stream from input table sql&quot;() {
          given:
          DataStream&lt;String&gt; streamRecords = env.readTextFile(&quot;/streaming_signals/pedals.txt&quot;)
          streaming.setConfiguration(configuration)
          streaming.setTableEnvironment(tableEnvironment);
      
          when:
          String tableSpec = streaming.createTableSpec(streamSpec);
          DataStream&lt;Row&gt; rawStream = streaming.getFilteredStream(streamSpec, tableSpec)
          DataStream&lt;String&gt; comapareStreams = rawStream.map(new MapFunction&lt;Row, String&gt;() {
              @Override
              public String map(Row record) {
                  // Logic to compare stream received with test stream
              }
          });
      
          then:
          // Comparison Logic
  
      }
  }
  </code></pre>
  <pre class="lang-sh prettyprint-override"><code>org.apache.flink.table.api.ValidationException: Could not find any factories that implement 'org.apache.flink.table.delegation.ExecutorFactory' in the classpath.
  at org.apache.flink.table.factories.FactoryUtil.discoverFactory(FactoryUtil.java:385)
  at org.apache.flink.table.api.internal.TableEnvironmentImpl.create(TableEnvironmentImpl.java:295)
  at org.apache.flink.table.api.internal.TableEnvironmentImpl.create(TableEnvironmentImpl.java:266)
  at org.apache.flink.table.api.TableEnvironment.create(TableEnvironment.java:95)
  at com.streaming.DataStreamTests.$spock_initializeFields(DataStreamTests.groovy:38
  </code></pre>