Unit testing filter function in flink

[Vishwas Siravara]

Hi guys,

I want to test a function like : 

private[flink] def filterStream(dataStream: DataStream[GenericRecord]): DataStream[GenericRecord] = {
  dataStream.filter(new FilterFunction[GenericRecord] {
    override def filter(value: GenericRecord): Boolean = {
      if (value == null || value.get(StipFields.requestMessageType) == null) {
        return false;
      } else {
        ExecutionEnv.messageTypeList.contains(value.get(StipFields.requestMessageType)
          .toString) && ExecutionEnv.pcrList.contains(value.get(StipFields.pcrList).toString) && (value.get(StipFields
          .rejectCode).asInstanceOf[Int] == 0) && !(value.get(StipFields.processingCode).toString.equals("33"))
      }
    }
  })
}

How can I do this ? 

Best,

Vishwas 

[vino yang]

Hi Vishwas,

Apache Flink provides some test harness to test your application code on multiple levels of the testing pyramid.

You can use them to test your UDF. Please see more examples offered by the official documentation[1].

Best,

Vino

[1]: https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/testing.html

Vishwas Siravara <[hidden email]> 于2019年12月20日周五 上午6:27写道：

Hi guys,

I want to test a function like : 

private[flink] def filterStream(dataStream: DataStream[GenericRecord]): DataStream[GenericRecord] = {
  dataStream.filter(new FilterFunction[GenericRecord] {
    override def filter(value: GenericRecord): Boolean = {
      if (value == null || value.get(StipFields.requestMessageType) == null) {
        return false;
      } else {
        ExecutionEnv.messageTypeList.contains(value.get(StipFields.requestMessageType)
          .toString) && ExecutionEnv.pcrList.contains(value.get(StipFields.pcrList).toString) && (value.get(StipFields
          .rejectCode).asInstanceOf[Int] == 0) && !(value.get(StipFields.processingCode).toString.equals("33"))
      }
    }
  })
}

How can I do this ? 

Best,

Vishwas