Apache Flinkアプリケーションのテスト戦略について、ステートレス/ステートフル関数からジョブレベルのテストまで網羅的に解説する。

1. テスト環境のセットアップ

Maven依存関係の設定例:

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.12</artifactId>
  <version>${flink.version}</version>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.mockito</groupId>
  <artifactId>mockito-core</artifactId>
  <version>4.0.0</version>
  <scope>test</scope>
</dependency>

2. ステートレス関数のテスト

2.1 MapFunctionのテスト例

public class IncrementMapper implements MapFunction<Long, Long> {
  @Override
  public Long map(Long input) {
    return input + 1;
  }
}

@Test
public void testMapper() {
  IncrementMapper mapper = new IncrementMapper();
  assertEquals(3L, (long)mapper.map(2L));
}

2.2 FlatMapFunctionのテスト例

public class Tokenizer implements FlatMapFunction<String, String> {
  @Override
  public void flatMap(String value, Collector<String> out) {
    for (String token : value.split(" ")) {
      out.collect(token);
    }
  }
}

@Test
public void testTokenizer() {
  Tokenizer tokenizer = new Tokenizer();
  Collector<String> mockCollector = mock(Collector.class);
  
  tokenizer.flatMap("test flink job", mockCollector);
  verify(mockCollector).collect("test");
  verify(mockCollector).collect("flink");
  verify(mockCollector).collect("job");
}

3. ステートフル関数のテスト

OperatorTestHarnessを使用したテスト例:

public class StatefulCounter extends RichFlatMapFunction<Long, Long> {
  private ValueState<Long> counter;

  @Override
  public void open(Configuration parameters) {
    counter = getRuntimeContext().getState(new ValueStateDescriptor<>("counter", Long.class));
  }

  @Override
  public void flatMap(Long value, Collector<Long> out) throws Exception {
    Long current = counter.value() == null ? 0L : counter.value();
    counter.update(current + value);
    out.collect(current + value);
  }
}

@Test
public void testStatefulOperator() throws Exception {
  OneInputStreamOperatorTestHarness<Long, Long> harness = 
    new OneInputStreamOperatorTestHarness<>(new StreamFlatMap<>(new StatefulCounter()));
  
  harness.open();
  harness.processElement(10L, 100L);
  harness.processElement(20L, 200L);
  
  assertEquals(Arrays.asList(10L, 30L), harness.extractOutputValues());
}

4. ジョブレベルの統合テスト

@ClassRule
public static MiniClusterWithClientResource cluster = new MiniClusterWithClientResource(
  new MiniClusterResourceConfiguration.Builder()
    .setNumberTaskManagers(1)
    .setNumberSlotsPerTaskManager(2)
    .build());

@Test
public void testCompletePipeline() throws Exception {
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  env.fromElements(1, 2, 3)
     .map(x -> x * 2)
     .addSink(new TestSink());
     
  env.execute();
  
  assertTrue(TestSink.results.containsAll(Arrays.asList(2, 4, 6)));
}

5. 時間処理のテスト

public class EventTimeProcessor extends ProcessFunction<Long, Long> {
  @Override
  public void processElement(Long value, Context ctx, Collector<Long> out) {
    ctx.timerService().registerEventTimeTimer(value + 1000);
  }

  @Override
  public void onTimer(long timestamp, OnTimerContext ctx, Collector<Long> out) {
    out.collect(timestamp);
  }
}

@Test
public void testEventTimeProcessing() throws Exception {
  OneInputStreamOperatorTestHarness<Long, Long> harness = 
    ProcessFunctionTestHarnesses.forKeyedProcessFunction(
      new EventTimeProcessor(), 
      x -> 1L, 
      Types.LONG);
      
  harness.processElement(100L, 100L);
  harness.processWatermark(1100L);
  
  assertEquals(Collections.singletonList(1100L), harness.extractOutputValues());
}