0
# Apache Flink Tests
1

2
Apache Flink Tests is a comprehensive integration test suite providing reusable testing infrastructure, utilities, and base classes for validating Apache Flink's stream processing functionality. It contains 50+ test utilities, base classes, and components covering fault tolerance, checkpointing, state migration, data generation, and streaming operations.
3

4
## Package Information
5

6
- **Package Name**: flink-tests_2.10
7
- **Package Type**: maven
8
- **Language**: Java (with Scala 2.10 support) 
9
- **Maven Coordinates**: `org.apache.flink:flink-tests_2.10:1.3.3`
10
- **Installation**: Add to your `pom.xml` dependencies with `<scope>test</scope>`
11

12
```xml
13
<dependency>
14
    <groupId>org.apache.flink</groupId>
15
    <artifactId>flink-tests_2.10</artifactId>
16
    <version>1.3.3</version>
17
    <scope>test</scope>
18
</dependency>
19
```
20

21
## Core Imports
22

23
```java
24
// Test utilities
25
import org.apache.flink.test.util.TestUtils;
26
import org.apache.flink.test.util.SuccessException;
27

28
// Data generation utilities
29
import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
30
import org.apache.flink.test.javaApiOperators.util.ValueCollectionDataSets;
31

32
// Base test classes for checkpointing and state management
33
import org.apache.flink.test.checkpointing.utils.SavepointMigrationTestBase;
34
import org.apache.flink.test.checkpointing.StreamFaultToleranceTestBase;
35

36
// Streaming test utilities
37
import org.apache.flink.test.streaming.runtime.util.EvenOddOutputSelector;
38
import org.apache.flink.test.streaming.runtime.util.TestListResultSink;
39
```
40

41
## Basic Usage
42

43
```java
44
// Using test data generation
45
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
46
DataSet<Tuple3<Integer, Long, String>> testData = 
47
    CollectionDataSets.get3TupleDataSet(env);
48

49
// Using test execution utilities
50
@Test
51
public void testStreamingJob() throws Exception {
52
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
53
    
54
    // Configure your streaming job
55
    env.fromElements(1, 2, 3, 4, 5)
56
       .map(x -> x * 2)
57
       .addSink(new TestListResultSink<>());
58
    
59
    // Execute with proper exception handling
60
    TestUtils.tryExecute(env, "Test Job");
61
}
62

63
// Using savepoint migration testing
64
public class MySavepointMigrationTest extends SavepointMigrationTestBase {
65
    @Test
66
    public void testMigration() throws Exception {
67
        // Test savepoint compatibility across versions
68
        StreamExecutionEnvironment env1 = createTestJob();
69
        executeAndSavepoint(env1, "test-savepoint");
70
        
71
        StreamExecutionEnvironment env2 = createModifiedJob();
72
        restoreAndExecute(env2, "test-savepoint");
73
    }
74
}
75
```
76

77
## Architecture
78

79
Apache Flink Tests is organized around several key infrastructure areas:
80

81
- **Core Test Utilities**: Execution helpers and exception handling for test scenarios
82
- **Data Generation Framework**: Comprehensive datasets and input formats for consistent testing
83
- **Checkpointing Infrastructure**: Base classes and utilities for testing savepoint migration and state compatibility
84
- **Fault Tolerance Testing**: Frameworks for testing recovery scenarios and failure injection
85
- **Streaming Test Components**: Specialized utilities for streaming operations and result collection
86
- **API Completeness Verification**: Tools for ensuring API parity between Java and Scala implementations
87

88
## Capabilities
89

90
### Test Data Generation
91

92
Comprehensive test data generation utilities providing consistent datasets for Java and Scala API testing. Includes custom types, collections, and input formats.
93

94
```java { .api }
95
public class CollectionDataSets {
96
    public static DataSet<Tuple3<Integer, Long, String>> get3TupleDataSet(ExecutionEnvironment env);
97
    public static DataSet<Tuple3<Integer, Long, String>> getSmall3TupleDataSet(ExecutionEnvironment env);
98
    public static DataSet<Tuple5<Integer, Long, Integer, String, Long>> get5TupleDataSet(ExecutionEnvironment env);
99
    public static DataSet<String> getStringDataSet(ExecutionEnvironment env);
100
    public static DataSet<Integer> getIntegerDataSet(ExecutionEnvironment env);
101
    public static DataSet<CustomType> getCustomTypeDataSet(ExecutionEnvironment env);
102
}
103
```
104

105
[Data Generation](./data-generation.md)
106

107
### Checkpointing and State Management
108

109
Infrastructure for testing savepoint migration, state compatibility, and checkpoint recovery across Flink versions. Essential for validating upgrade paths and state serialization.
110

111
```java { .api }
112
public abstract class SavepointMigrationTestBase {
113
    protected void executeAndSavepoint(JobGraph jobGraph, String savepointName) throws Exception;
114
    protected void restoreAndExecute(JobGraph jobGraph, String savepointName) throws Exception;
115
    protected abstract String getResourceFilename(String filename);
116
}
117
```
118

119
[Checkpointing and State Management](./checkpointing.md)
120

121
### Fault Tolerance Testing
122

123
Base classes and utilities for testing job cancellation, task failure recovery, and fault tolerance mechanisms. Provides controlled failure injection and recovery validation.
124

125
```java { .api }
126
public abstract class StreamFaultToleranceTestBase {
127
    protected abstract void testProgram(StreamExecutionEnvironment env);
128
    protected void postSubmit() throws Exception;
129
}
130

131
public abstract class SimpleRecoveryITCaseBase {
132
    protected void execute() throws Exception;
133
}
134
```
135

136
[Fault Tolerance](./fault-tolerance.md)
137

138
### Streaming Test Utilities
139

140
Specialized components for testing streaming operations including output selectors, result collection, and stream partitioning utilities.
141

142
```java { .api }
143
public class EvenOddOutputSelector implements OutputSelector<Integer> {
144
    public Iterable<String> select(Integer value);
145
}
146

147
public class TestListResultSink<T> implements SinkFunction<T> {
148
    public void invoke(T value);
149
    public List<T> getResult();
150
}
151
```
152

153
[Streaming Utilities](./streaming.md)
154

155
### Test Execution Utilities
156

157
Core utilities for executing test jobs with proper exception handling and result validation. Handles the complexities of test execution in Flink environments.
158

159
```java { .api }
160
public class TestUtils {
161
    public static JobExecutionResult tryExecute(StreamExecutionEnvironment env, String jobName) throws Exception;
162
}
163

164
public class SuccessException extends Exception {
165
    public SuccessException();
166
}
167
```
168

169
[Execution Utilities](./execution.md)
170

171
### API Completeness Testing
172

173
Framework for verifying API parity between Java and Scala implementations using reflection-based method comparison.
174

175
```java { .api }
176
public abstract class ScalaAPICompletenessTestBase {
177
    protected void compareApis(Class<?> javaClass, Class<?> scalaClass);
178
    protected boolean isExcluded(Method method);
179
}
180
```
181

182
[API Completeness](./api-completeness.md)