0
# Flink DSTL DFS
1
2
Flink DSTL DFS provides a distributed file system-based implementation of changelog storage for Apache Flink's streaming state backend. It enables efficient state change tracking and recovery by persisting state modifications to distributed file systems like HDFS or S3. The library supports fault-tolerance guarantees by providing reliable state change persistence with built-in metrics, configurable upload scheduling, and preemptive persistence optimizations.
3
4
## Package Information
5
6
- **Package Name**: flink-dstl-dfs
7
- **Package Type**: maven
8
- **Language**: Java
9
- **Group ID**: org.apache.flink
10
- **Artifact ID**: flink-dstl-dfs
11
- **Installation**: Add to Maven dependencies:
12
13
```xml
14
<dependency>
15
<groupId>org.apache.flink</groupId>
16
<artifactId>flink-dstl-dfs</artifactId>
17
<version>2.1.0</version>
18
</dependency>
19
```
20
21
## Core Imports
22
23
```java
24
import org.apache.flink.changelog.fs.FsStateChangelogStorageFactory;
25
import org.apache.flink.changelog.fs.FsStateChangelogStorage;
26
import org.apache.flink.changelog.fs.FsStateChangelogOptions;
27
```
28
29
## Basic Usage
30
31
```java
32
import org.apache.flink.api.common.JobID;
33
import org.apache.flink.changelog.fs.FsStateChangelogStorageFactory;
34
import org.apache.flink.configuration.Configuration;
35
import org.apache.flink.runtime.state.LocalRecoveryConfig;
36
import org.apache.flink.runtime.metrics.groups.TaskManagerJobMetricGroup;
37
38
// Configure filesystem changelog storage
39
Configuration config = new Configuration();
40
FsStateChangelogStorageFactory.configure(
41
config,
42
new File("/path/to/changelog/storage"),
43
Duration.ofSeconds(10), // upload timeout
44
3 // max retry attempts
45
);
46
47
// Create storage factory
48
FsStateChangelogStorageFactory factory = new FsStateChangelogStorageFactory();
49
50
// Create storage instance
51
FsStateChangelogStorage storage = (FsStateChangelogStorage) factory.createStorage(
52
new JobID(),
53
config,
54
taskManagerJobMetricGroup,
55
localRecoveryConfig
56
);
57
58
// Create changelog writer for an operator
59
FsStateChangelogWriter writer = storage.createWriter(
60
"operator-id",
61
KeyGroupRange.of(0, 127),
62
mailboxExecutor
63
);
64
65
// Append state changes
66
writer.append(0, stateChangeBytes);
67
writer.appendMeta(metadataBytes);
68
69
// Persist changes during checkpoint
70
CompletableFuture<SnapshotResult<ChangelogStateHandleStreamImpl>> result =
71
writer.persist(sequenceNumber, checkpointId);
72
```
73
74
## Architecture
75
76
The Flink DSTL DFS library is organized around several key components:
77
78
- **Storage Factory**: `FsStateChangelogStorageFactory` creates storage instances and provides service loader integration
79
- **Storage Implementation**: `FsStateChangelogStorage` manages changelog writers and upload scheduling
80
- **Writers**: `FsStateChangelogWriter` handles state change appending and persistence operations
81
- **Upload System**: Configurable upload schedulers handle batching and asynchronous persistence to distributed file systems
82
- **Recovery System**: `FsStateChangelogStorageForRecovery` provides read-only access for checkpoint recovery
83
- **Configuration**: Comprehensive options for tuning performance, retry policies, and storage behavior
84
85
## Capabilities
86
87
### Storage Factory and Configuration
88
89
Factory for creating filesystem-based changelog storage instances with comprehensive configuration support for distributed file systems.
90
91
```java { .api }
92
public class FsStateChangelogStorageFactory implements StateChangelogStorageFactory {
93
public static final String IDENTIFIER = "filesystem";
94
95
public String getIdentifier();
96
public StateChangelogStorage<?> createStorage(
97
JobID jobID,
98
Configuration configuration,
99
TaskManagerJobMetricGroup metricGroup,
100
LocalRecoveryConfig localRecoveryConfig
101
) throws IOException;
102
public StateChangelogStorageView<?> createStorageView(Configuration configuration);
103
public static void configure(
104
Configuration configuration,
105
File newFolder,
106
Duration uploadTimeout,
107
int maxUploadAttempts
108
);
109
}
110
```
111
112
[Storage Factory and Configuration](./storage-factory.md)
113
114
### Main Storage Implementation
115
116
Core filesystem-based implementation providing changelog writers and managing upload operations for high-throughput streaming applications.
117
118
```java { .api }
119
public class FsStateChangelogStorage extends FsStateChangelogStorageForRecovery
120
implements StateChangelogStorage<ChangelogStateHandleStreamImpl> {
121
122
public FsStateChangelogWriter createWriter(
123
String operatorID,
124
KeyGroupRange keyGroupRange,
125
MailboxExecutor mailboxExecutor
126
);
127
public void close() throws Exception;
128
public AvailabilityProvider getAvailabilityProvider();
129
}
130
```
131
132
[Main Storage Implementation](./storage-implementation.md)
133
134
### Changelog Writers
135
136
Writers for appending state changes and managing persistence operations with preemptive flushing and checkpoint coordination.
137
138
```java { .api }
139
interface StateChangelogWriter<T> {
140
void append(int keyGroup, byte[] value) throws IOException;
141
void appendMeta(byte[] value) throws IOException;
142
SequenceNumber nextSequenceNumber();
143
CompletableFuture<SnapshotResult<T>> persist(SequenceNumber from, long checkpointId) throws IOException;
144
void truncate(SequenceNumber to);
145
void truncateAndClose(SequenceNumber from);
146
void confirm(SequenceNumber from, SequenceNumber to, long checkpointId);
147
void reset(SequenceNumber from, SequenceNumber to, long checkpointId);
148
void close();
149
}
150
```
151
152
[Changelog Writers](./changelog-writers.md)
153
154
### Configuration Options
155
156
Comprehensive configuration options for performance tuning, retry policies, and storage behavior optimization.
157
158
```java { .api }
159
public class FsStateChangelogOptions {
160
public static final ConfigOption<String> BASE_PATH;
161
public static final ConfigOption<Boolean> COMPRESSION_ENABLED;
162
public static final ConfigOption<MemorySize> PREEMPTIVE_PERSIST_THRESHOLD;
163
public static final ConfigOption<Duration> PERSIST_DELAY;
164
public static final ConfigOption<MemorySize> PERSIST_SIZE_THRESHOLD;
165
public static final ConfigOption<MemorySize> UPLOAD_BUFFER_SIZE;
166
public static final ConfigOption<Integer> NUM_UPLOAD_THREADS;
167
public static final ConfigOption<MemorySize> IN_FLIGHT_DATA_LIMIT;
168
public static final ConfigOption<String> RETRY_POLICY;
169
public static final ConfigOption<Duration> UPLOAD_TIMEOUT;
170
public static final ConfigOption<Integer> RETRY_MAX_ATTEMPTS;
171
}
172
```
173
174
[Configuration Options](./configuration-options.md)
175
176
### Upload Scheduling and Management
177
178
Upload scheduling system with batching, throttling, and retry capabilities for efficient distributed file system operations.
179
180
```java { .api }
181
public interface StateChangeUploadScheduler extends AutoCloseable {
182
void upload(UploadTask task);
183
AvailabilityProvider getAvailabilityProvider();
184
void close() throws Exception;
185
186
static StateChangeUploadScheduler fromConfig(
187
JobID jobID,
188
Configuration config,
189
ChangelogStorageMetricGroup metricGroup,
190
TaskChangelogRegistry changelogRegistry,
191
LocalRecoveryConfig localRecoveryConfig
192
) throws IOException;
193
}
194
195
public interface StateChangeUploader extends AutoCloseable {
196
UploadTasksResult upload(Collection<UploadTask> tasks) throws IOException;
197
}
198
```
199
200
[Upload System](./upload-system.md)
201
202
### Recovery and State Management
203
204
Recovery system providing read-only access to persisted changelog data and lifecycle management for state handles.
205
206
```java { .api }
207
public class FsStateChangelogStorageForRecovery
208
implements StateChangelogStorageView<ChangelogStateHandleStreamImpl> {
209
210
public StateChangelogHandleReader<ChangelogStateHandleStreamImpl> createReader();
211
public void close() throws Exception;
212
}
213
214
public interface TaskChangelogRegistry {
215
void startTracking(StreamStateHandle handle, long refCount);
216
void stopTracking(StreamStateHandle handle);
217
void release(StreamStateHandle handle);
218
}
219
```
220
221
[Recovery and State Management](./recovery-system.md)
222
223
### Metrics and Monitoring
224
225
Comprehensive metrics collection for monitoring upload performance, failure rates, and system health in production environments.
226
227
```java { .api }
228
public class ChangelogStorageMetricGroup extends ProxyMetricGroup<MetricGroup> {
229
// Provides counters for uploads, failures, batch sizes, latencies, and retry attempts
230
}
231
```
232
233
[Metrics and Monitoring](./metrics-monitoring.md)
234
235
## Types
236
237
```java { .api }
238
public final class UploadResult {
239
public final StreamStateHandle streamStateHandle;
240
public final StreamStateHandle localStreamHandle;
241
public final long offset;
242
public final long localOffset;
243
public final SequenceNumber sequenceNumber;
244
public final long size;
245
246
public UploadResult(StreamStateHandle streamStateHandle, long offset,
247
SequenceNumber sequenceNumber, long size);
248
public StreamStateHandle getStreamStateHandle();
249
public long getOffset();
250
public SequenceNumber getSequenceNumber();
251
public long getSize();
252
}
253
254
public class StateChangeSet {
255
public StateChangeSet(UUID logId, SequenceNumber sequenceNumber, List<StateChange> changes);
256
public UUID getLogId();
257
public List<StateChange> getChanges();
258
public SequenceNumber getSequenceNumber();
259
public long getSize();
260
}
261
262
public interface RetryPolicy {
263
long timeoutFor(int attempt);
264
long retryAfter(int failedAttempt, Exception exception);
265
266
static RetryPolicy fromConfig(ReadableConfig config);
267
static RetryPolicy fixed(int maxAttempts, long timeout, long delayAfterFailure);
268
RetryPolicy NONE = /* ... */;
269
}
270
```