테스팅

테스팅 (Testing)

테스팅은 모든 소프트웨어 개발 프로세스의 필수적인 부분이에요. Apache Flink는 테스팅 피라미드의 여러 수준에서 애플리케이션 코드를 테스트할 수 있는 도구를 제공해요.

출처: 문서

본문

사용자 정의 함수 테스팅 (Testing User-Defined Functions)

보통 Flink가 사용자 정의 함수 밖에서는 올바른 결과를 만든다고 가정할 수 있어요. 따라서 주요 비즈니스 로직을 포함하는 클래스는 가능한 한 단위 테스트로 테스트하는 것을 권장해요.

상태 없는(Stateless), 무시간(timeless) UDF 단위 테스트

예를 들어 다음 상태 없는 MapFunction을 봅시다.

public class IncrementMapFunction implements MapFunction<Long, Long> {

    @Override
    public Long map(Long record) throws Exception {
        return record + 1;
    }
}

즐겨 쓰는 테스팅 프레임워크로 적절한 인자를 전달하고 출력을 검증해 그러한 함수를 단위 테스트하는 것은 매우 쉬워요.

public class IncrementMapFunctionTest {

    @Test
    public void testIncrement() throws Exception {
        // instantiate your function
        IncrementMapFunction incrementer = new IncrementMapFunction();

        // call the methods that you have implemented
        assertEquals(3L, incrementer.map(2L));
    }
}

마찬가지로 org.apache.flink.util.Collector를 사용하는 사용자 정의 함수(예: FlatMapFunction 또는 ProcessFunction)는 실제 컬렉터 대신 mock 객체를 제공해 쉽게 테스트할 수 있어요. IncrementMapFunction과 같은 기능을 가진 FlatMapFunction은 다음과 같이 단위 테스트될 수 있어요.

public class IncrementFlatMapFunctionTest {

    @Test
    public void testIncrement() throws Exception {
        // instantiate your function
        IncrementFlatMapFunction incrementer = new IncrementFlatMapFunction();

        Collector<Integer> collector = mock(Collector.class);

        // call the methods that you have implemented
        incrementer.flatMap(2L, collector);

        //verify collector was called with the right output
        Mockito.verify(collector, times(1)).collect(3L);
    }
}

상태 있는(Stateful) 또는 시기적절(timely) UDF 및 사용자 정의 연산자 단위 테스트

관리되는 상태(managed state)나 타이머를 사용하는 사용자 정의 함수의 기능을 테스트하는 것은, 사용자 코드와 Flink 런타임 사이의 상호작용을 테스트하는 것을 포함하므로 더 어려워요.

이를 위해 Flink는 소위 test harnesses 컬렉션을 제공하며, 이를 사용해 그러한 사용자 정의 함수와 사용자 정의 연산자를 테스트할 수 있어요.

  • OneInputStreamOperatorTestHarness (DataStream의 연산자용)
  • KeyedOneInputStreamOperatorTestHarness (KeyedStream의 연산자용)
  • TwoInputStreamOperatorTestHarness (두 DataStream의 ConnectedStream 연산자용)
  • KeyedTwoInputStreamOperatorTestHarness (두 KeyedStream의 ConnectedStream 연산자용)

test harness를 사용하려면 추가 의존성 집합이 필요해요. 자세한 내용은 구성 절을 참고하세요.

이제 test harness를 사용해 레코드와 워터마크를 사용자 정의 함수나 사용자 정의 연산자에 밀어 넣고, 처리 시간을 제어하고, 마지막으로 연산자의 출력(사이드 출력 포함)을 검증할 수 있어요.

public class StatefulFlatMapTest {
    private OneInputStreamOperatorTestHarness<Long, Long> testHarness;
    private StatefulFlatMap statefulFlatMapFunction;

    @Before
    public void setupTestHarness() throws Exception {

        //instantiate user-defined function
        statefulFlatMapFunction = new StatefulFlatMapFunction();

        // wrap user defined function into a the corresponding operator
        testHarness = new OneInputStreamOperatorTestHarness<>(new StreamFlatMap<>(statefulFlatMapFunction));

        // optionally configured the execution environment
        testHarness.getExecutionConfig().setAutoWatermarkInterval(50);

        // open the test harness (will also call open() on RichFunctions)
        testHarness.open();
    }

    @Test
    public void testingStatefulFlatMapFunction() throws Exception {

        //push (timestamped) elements into the operator (and hence user defined function)
        testHarness.processElement(2L, 100L);

        //trigger event time timers by advancing the event time of the operator with a watermark
        testHarness.processWatermark(100L);

        //trigger processing time timers by advancing the processing time of the operator directly
        testHarness.setProcessingTime(100L);

        //retrieve list of emitted records for assertions
        assertThat(testHarness.getOutput(), containsInExactlyThisOrder(3L));

        //retrieve list of records emitted to a specific side output for assertions (ProcessFunction only)
        //assertThat(testHarness.getSideOutput(new OutputTag<>("invalidRecords")), hasSize(0))
    }
}

KeyedOneInputStreamOperatorTestHarness와 KeyedTwoInputStreamOperatorTestHarness는 키 클래스의 TypeInformation을 포함한 KeySelector를 추가로 제공해 인스턴스화돼요.

public class StatefulFlatMapFunctionTest {
    private OneInputStreamOperatorTestHarness<String, Long, Long> testHarness;
    private StatefulFlatMap statefulFlatMapFunction;

    @Before
    public void setupTestHarness() throws Exception {

        //instantiate user-defined function
        statefulFlatMapFunction = new StatefulFlatMapFunction();

        // wrap user defined function into a the corresponding operator
        testHarness = new KeyedOneInputStreamOperatorTestHarness<>(new StreamFlatMap<>(statefulFlatMapFunction), new MyStringKeySelector(), Types.STRING);

        // open the test harness (will also call open() on RichFunctions)
        testHarness.open();
    }

    //tests

}

이 test harness 사용법의 많은 예시는 Flink 코드베이스에서 찾을 수 있어요. 예를 들어:

  • org.apache.flink.streaming.runtime.operators.windowing.WindowOperatorTest는 처리 시간 또는 이벤트 시간에 의존하는 연산자와 사용자 정의 함수를 테스트하는 좋은 예시예요.

참고: AbstractStreamOperatorTestHarness와 그 파생 클래스는 현재 공개 API의 일부가 아니며 변경될 수 있다는 점에 유의하세요.

ProcessFunction 단위 테스트 (Unit Testing ProcessFunction)

그 중요성을 감안해, ProcessFunction을 직접 테스트하는 데 사용할 수 있는 이전 test harness들에 더해, Flink는 ProcessFunctionTestHarnesses라는 test harness 팩토리를 제공해 test harness 인스턴스화를 더 쉽게 만들어요. 이 예시를 고려해 봅시다.

참고: 이 test harness를 사용하려면 마지막 절에서 언급한 의존성도 도입해야 해요.

public static class PassThroughProcessFunction extends ProcessFunction<Integer, Integer> {

	@Override
	public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
        out.collect(value);
	}
}

적절한 인자를 전달하고 출력을 검증해 ProcessFunctionTestHarnesses로 그러한 함수를 단위 테스트하는 것은 매우 쉬워요.

public class PassThroughProcessFunctionTest {

    @Test
    public void testPassThrough() throws Exception {

        //instantiate user-defined function
        PassThroughProcessFunction processFunction = new PassThroughProcessFunction();

        // wrap user defined function into a the corresponding operator
        OneInputStreamOperatorTestHarness<Integer, Integer> harness = ProcessFunctionTestHarnesses
        	.forProcessFunction(processFunction);

        //push (timestamped) elements into the operator (and hence user defined function)
        harness.processElement(1, 10);

        //retrieve list of emitted records for assertions
        assertEquals(harness.extractOutputValues(), Collections.singletonList(1));
    }
}

KeyedProcessFunction, KeyedCoProcessFunction, BroadcastProcessFunction 등 ProcessFunction의 다양한 변형을 테스트하기 위해 ProcessFunctionTestHarnesses를 사용하는 방법에 대한 더 많은 예시를 보려면 ProcessFunctionTestHarnessesTest를 살펴보는 것이 좋아요.

JUnit Rule MiniClusterWithClientResource

Apache Flink는 로컬 임베디드 미니 클러스터에 대해 전체 job을 테스트하기 위한 JUnit rule MiniClusterWithClientResource를 제공해요. MiniClusterWithClientResource를 사용하려면 추가 의존성(테스트 스코프)이 하나 필요해요.

    org.apache.flink
    flink-test-utils
    2.3.0    
    test

이전 절들과 같은 단순한 MapFunction을 봅시다.

public class IncrementMapFunction implements MapFunction<Long, Long> {

    @Override
    public Long map(Long record) throws Exception {
        return record + 1;
    }
}

이 MapFunction을 사용하는 간단한 파이프라인은 다음과 같이 로컬 Flink 클러스터에서 테스트할 수 있어요.

public class ExampleIntegrationTest {

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

    @Test
    public void testIncrementPipeline() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // configure your test environment
        env.setParallelism(2);

        // values are collected in a static variable
        CollectSink.values.clear();

        // create a stream of custom elements and apply transformations
        env.fromElements(1L, 21L, 22L)
                .map(new IncrementMapFunction())
                .addSink(new CollectSink());

        // execute
        env.execute();

        // verify your results
        assertTrue(CollectSink.values.containsAll(2L, 22L, 23L));
    }

    // create a testing sink
    private static class CollectSink implements SinkFunction<Long> {

        // must be static
        public static final List<Long> values = Collections.synchronizedList(new ArrayList<>());

        @Override
        public void invoke(Long value, SinkFunction.Context context) throws Exception {
            values.add(value);
        }
    }
}

MiniClusterWithClientResource로 통합 테스트하는 데 대한 몇 가지 언급:

  • 프로덕션에서 테스트로 파이프라인 코드 전체를 복사하지 않기 위해, 프로덕션 코드에서 소스와 싱크를 플러그 가능하게 만들고 테스트에서 특별한 테스트 소스와 테스트 싱크를 주입해요.
  • CollectSink의 정적 변수는 Flink가 클러스터에 분배하기 전에 모든 연산자를 직렬화하기 때문에 여기서 사용돼요. 로컬 Flink 미니 클러스터가 인스턴스화한 연산자와 정적 변수를 통해 통신하는 것은 이 문제를 해결하는 한 가지 방법이에요. 대안으로 테스트 싱크로 임시 디렉터리의 파일에 데이터를 쓸 수도 있어요.
  • job이 이벤트 시간 타이머를 사용한다면 워터마크를 방출하기 위해 사용자 정의 병렬 소스 함수를 구현할 수 있어요.
  • 병렬로 실행되는 파이프라인에서만 나타나는 버그를 식별하기 위해, 항상 parallelism > 1로 파이프라인을 로컬에서 테스트하는 것을 권장해요.
  • @Rule보다 @ClassRule을 선호해서 여러 테스트가 같은 Flink 클러스터를 공유하게 해요. Flink 클러스터의 시작·종료가 실제 테스트의 실행 시간을 지배하므로, 이는 상당한 시간을 절약해요.
  • 파이프라인에 사용자 정의 상태 처리가 포함되어 있다면, 미니 클러스터 안에서 체크포인팅을 활성화하고 job을 재시작해 정확성을 테스트할 수 있어요. 이를 위해 파이프라인의 (테스트 전용) 사용자 정의 함수에서 예외를 던져 실패를 트리거해야 해요.

더 알아보기 (Learn more)