java - Flink 函数单元测试

Question

我使用的是 Flink v.1.4.0。

我已经实现了一个模块，作为我正在开发的包的一部分，其作用是对流进行重复数据删除。该模块非常简单:

public class RemoveDuplicateFilter<T> extends RichFlatMapFunction<T, T> {

static final ValueStateDescriptor<Boolean> SEEN_DESCRIPTOR = new ValueStateDescriptor<>("seen", Boolean.class);
private ValueState<Boolean> seen;

@Override
public void open(Configuration configuration) {
  RuntimeContext runtimeContext = this.getRuntimeContext();
  seen = runtimeContext.getState(SEEN_DESCRIPTOR);  
}

@Override
public void flatMap(T value, Collector<T> out) throws Exception {
  Boolean hasBeenSeen = seen.value();

  if(hasBeenSeen == null || !hasBeenSeen) {
    out.collect(value);
    seen.update(true);
  }  
}

问题是:如何在无需实例化实际 Flink ValueState 的情况下测试此代码？即使用 Mockito？

我尝试了很多方法，但本质上，当涉及到调用时:

RuntimeContext runtimeContext = Mockito.mock(RuntimeContext.class);
...
when(runtimeContext.getState(SEEN_DESCRIPTOR)).thenReturn(seen);

调用总是失败。我尝试用 Matchers.any() 替换 SEEN_DESCRIPTOR 但仍然没有成功。

有什么建议吗？

Answer 1

您可以使用flinkspector对功能进行单元测试。