org.apache.flink.connector.testframe.testsuites

Class SinkTestSuiteBase<T extends Comparable<T>>

java.lang.Object

org.apache.flink.connector.testframe.testsuites.SinkTestSuiteBase<T>

@ExtendWith(value={ConnectorTestingExtension.class,TestLoggerExtension.class,TestCaseInvocationContextProvider.class})
 @TestInstance(value=PER_CLASS)
 @Experimental
public abstract class SinkTestSuiteBase<T extends Comparable<T>>
extends Object

Base class for sink test suite.

All cases should have well-descriptive JavaDoc, including:

• What's the purpose of this case
• Simple description of how this case works
• Condition to fulfill in order to pass this case
• Requirement of running this case

Constructor Summary

Constructors

Constructor and Description
SinkTestSuiteBase()

Method Summary

Modifier and Type	Method and Description
protected CollectResultIterator<T>	addCollectSink(DataStream<T> stream)
protected void	checkResultWithSemantic(ExternalSystemDataReader<T> reader, List<T> testData, CheckpointingMode semantic) Compare the test data with actual data in given semantic.
protected List<T>	generateTestData(TestingSinkSettings testingSinkSettings, DataStreamSinkExternalContext<T> externalContext) Generate a set of test records.
void	testBasicSink(TestEnvironment testEnv, DataStreamSinkExternalContext<T> externalContext, CheckpointingMode semantic) Test DataStream connector sink.
void	testMetrics(TestEnvironment testEnv, DataStreamSinkExternalContext<T> externalContext, CheckpointingMode semantic) Test connector sink metrics.
void	testScaleDown(TestEnvironment testEnv, DataStreamSinkExternalContext<T> externalContext, CheckpointingMode semantic) Test connector sink restart from a completed savepoint with a lower parallelism.
void	testScaleUp(TestEnvironment testEnv, DataStreamSinkExternalContext<T> externalContext, CheckpointingMode semantic) Test connector sink restart from a completed savepoint with a higher parallelism.
void	testStartFromSavepoint(TestEnvironment testEnv, DataStreamSinkExternalContext<T> externalContext, CheckpointingMode semantic) Test connector sink restart from a completed savepoint with the same parallelism.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

SinkTestSuiteBase

public SinkTestSuiteBase()

Method Detail

testBasicSink

@TestTemplate
 @DisplayName(value="Test data stream sink")
public void testBasicSink(TestEnvironment testEnv,
                                                                                     DataStreamSinkExternalContext<T> externalContext,
                                                                                     CheckpointingMode semantic)
                                                                              throws Exception

Test DataStream connector sink.

The following tests will create a sink in the external system, generate a collection of test data and write them to this sink by the Flink Job.

In order to pass these tests, the number of records produced by Flink need to be equals to the generated test data. And the records in the sink will be compared to the test data by the different semantics. There's no requirement for records order.

Throws:

Exception

testStartFromSavepoint

@TestTemplate
 @DisplayName(value="Test sink restarting from a savepoint")
public void testStartFromSavepoint(TestEnvironment testEnv,
                                                                                                              DataStreamSinkExternalContext<T> externalContext,
                                                                                                              CheckpointingMode semantic)
                                                                                                       throws Exception

Test connector sink restart from a completed savepoint with the same parallelism.

This test will create a sink in the external system, generate a collection of test data and write a half part of them to this sink by the Flink Job with parallelism 2 at first. Then stop the job, restart the same job from the completed savepoint. After the job has been running, write the other part to the sink and compare the result.

In order to pass this test, the number of records produced by Flink need to be equals to the generated test data. And the records in the sink will be compared to the test data by the different semantic. There's no requirement for record order.

Throws:

Exception

testScaleUp

@TestTemplate
 @DisplayName(value="Test sink restarting with a higher parallelism")
public void testScaleUp(TestEnvironment testEnv,
                                                                                                            DataStreamSinkExternalContext<T> externalContext,
                                                                                                            CheckpointingMode semantic)
                                                                                                     throws Exception

Test connector sink restart from a completed savepoint with a higher parallelism.

This test will create a sink in the external system, generate a collection of test data and write a half part of them to this sink by the Flink Job with parallelism 2 at first. Then stop the job, restart the same job from the completed savepoint with a higher parallelism 4. After the job has been running, write the other part to the sink and compare the result.

In order to pass this test, the number of records produced by Flink need to be equals to the generated test data. And the records in the sink will be compared to the test data by the different semantic. There's no requirement for record order.

Throws:

Exception

testScaleDown

@TestTemplate
 @DisplayName(value="Test sink restarting with a lower parallelism")
public void testScaleDown(TestEnvironment testEnv,
                                                                                                             DataStreamSinkExternalContext<T> externalContext,
                                                                                                             CheckpointingMode semantic)
                                                                                                      throws Exception

Test connector sink restart from a completed savepoint with a lower parallelism.

This test will create a sink in the external system, generate a collection of test data and write a half part of them to this sink by the Flink Job with parallelism 4 at first. Then stop the job, restart the same job from the completed savepoint with a lower parallelism 2. After the job has been running, write the other part to the sink and compare the result.

In order to pass this test, the number of records produced by Flink need to be equals to the generated test data. And the records in the sink will be compared to the test data by the different semantic. There's no requirement for record order.

Throws:

Exception

testMetrics

@TestTemplate
 @DisplayName(value="Test sink metrics")
public void testMetrics(TestEnvironment testEnv,
                                                                               DataStreamSinkExternalContext<T> externalContext,
                                                                               CheckpointingMode semantic)
                                                                        throws Exception

Test connector sink metrics.

This test will create a sink in the external system, generate test data and write them to the sink via a Flink job. Then read and compare the metrics.

Now test: numRecordsOut

Throws:

Exception

generateTestData

protected List<T> generateTestData(TestingSinkSettings testingSinkSettings,
                                   DataStreamSinkExternalContext<T> externalContext)

Generate a set of test records.

Parameters:

testingSinkSettings - sink settings

externalContext - External context

Returns:

Collection of generated test records

checkResultWithSemantic

protected void checkResultWithSemantic(ExternalSystemDataReader<T> reader,
                                       List<T> testData,
                                       CheckpointingMode semantic)
                                throws Exception

Compare the test data with actual data in given semantic.

Parameters:

reader - the data reader for the sink

testData - the test data

semantic - the supported semantic, see CheckpointingMode

Throws:

Exception

addCollectSink

protected CollectResultIterator<T> addCollectSink(DataStream<T> stream)