0
# Apache Flink Clients
1

2
Apache Flink Clients provides comprehensive client-side functionality for interacting with Flink clusters. This library includes command-line interfaces, programmatic cluster clients, deployment management, application packaging, and artifact handling capabilities for both batch and stream processing workloads.
3

4
## Package Information
5

6
- **Package Name**: flink-clients
7
- **Package Type**: maven
8
- **Language**: Java
9
- **Group ID**: org.apache.flink
10
- **Artifact ID**: flink-clients
11
- **Version**: 2.1.0
12
- **Installation**: 
13
  ```xml
14
  <dependency>
15
    <groupId>org.apache.flink</groupId>
16
    <artifactId>flink-clients</artifactId>
17
    <version>2.1.0</version>
18
  </dependency>
19
  ```
20

21
## Core Imports
22

23
```java
24
// CLI Frontend for command-line operations
25
import org.apache.flink.client.cli.CliFrontend;
26

27
// Core cluster client interface
28
import org.apache.flink.client.program.ClusterClient;
29

30
// Program packaging and execution
31
import org.apache.flink.client.program.PackagedProgram;
32

33
// Deployment management
34
import org.apache.flink.client.deployment.ClusterDescriptor;
35
import org.apache.flink.client.deployment.ClusterClientFactory;
36

37
// Application mode deployment
38
import org.apache.flink.client.deployment.application.ApplicationRunner;
39
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
40
```
41

42
## Basic Usage
43

44
```java
45
import org.apache.flink.client.program.ClusterClient;
46
import org.apache.flink.client.program.PackagedProgram;
47
import org.apache.flink.client.program.rest.RestClusterClient;
48
import org.apache.flink.client.program.rest.RestClusterClientConfiguration;
49
import org.apache.flink.configuration.Configuration;
50
import org.apache.flink.configuration.RestOptions;
51

52
// Create cluster client configuration
53
Configuration config = new Configuration();
54
config.setString(RestOptions.ADDRESS, "localhost");
55
config.setInteger(RestOptions.PORT, 8081);
56

57
// Create REST cluster client
58
RestClusterClientConfiguration clientConfig = 
59
    RestClusterClientConfiguration.fromConfiguration(config);
60
try (ClusterClient<?> client = new RestClusterClient<>(clientConfig, "default")) {
61
    
62
    // Package and submit a job
63
    PackagedProgram program = PackagedProgram.newBuilder()
64
        .setJarFile(new File("/path/to/job.jar"))
65
        .setArguments(new String[]{"--input", "input.txt"})
66
        .build();
67
    
68
    // Submit job to cluster
69
    JobID jobId = ClientUtils.executeProgram(
70
        new DefaultExecutorServiceLoader(), 
71
        config, 
72
        program, 
73
        false, 
74
        false
75
    ).getJobID();
76
    
77
    System.out.println("Submitted job: " + jobId);
78
}
79
```
80

81
## Architecture
82

83
Apache Flink Clients is built around several key components:
84

85
- **CLI Frontend**: `CliFrontend` provides comprehensive command-line interface for job submission, management, and monitoring
86
- **Cluster Clients**: `ClusterClient` interface abstracts cluster communication with implementations for REST, standalone, and mini clusters
87
- **Program Packaging**: `PackagedProgram` handles JAR-based job packaging, classpath management, and main class discovery
88
- **Deployment Framework**: Pluggable deployment system supporting multiple cluster types (standalone, YARN, Kubernetes)
89
- **Application Mode**: Dedicated application deployment for long-running applications with `ApplicationRunner` and related components
90
- **Artifact Management**: Flexible artifact fetching system supporting local files, HTTP, and distributed file systems
91

92
## Capabilities
93

94
### Command-Line Interface
95

96
Complete CLI frontend for Flink operations including job submission, cancellation, savepoint management, and cluster monitoring.
97

98
```java { .api }
99
public class CliFrontend {
100
    public static void main(String[] args);
101
    public int parseAndRun(String[] args);
102
}
103
```
104

105
[Command-Line Interface](./cli-frontend.md)
106

107
### Cluster Client Management
108

109
Core interface for programmatic cluster interaction, supporting job submission, status monitoring, and cluster lifecycle management.
110

111
```java { .api }
112
public interface ClusterClient<T> extends AutoCloseable {
113
    T getClusterId();
114
    Configuration getFlinkConfiguration();
115
    CompletableFuture<Collection<JobStatusMessage>> listJobs() throws Exception;
116
    CompletableFuture<JobID> submitJob(ExecutionPlan executionPlan);
117
    CompletableFuture<JobStatus> getJobStatus(JobID jobId);
118
    String getWebInterfaceURL();
119
    void shutDownCluster();
120
}
121
```
122

123
[Cluster Client Management](./cluster-client.md)
124

125
### Program Packaging and Execution
126

127
Functionality for packaging Flink programs in JAR files, managing classpaths, and executing user applications.
128

129
```java { .api }
130
public class PackagedProgram implements AutoCloseable {
131
    public static Builder newBuilder();
132
    public Class<?> getMainClass() throws ProgramInvocationException;
133
    public String[] getArguments();
134
    public void invokeInteractiveModeForExecution() throws ProgramInvocationException;
135
}
136
```
137

138
[Program Packaging](./program-packaging.md)
139

140
### Deployment Management
141

142
Cluster deployment abstraction supporting multiple deployment targets with pluggable factory pattern.
143

144
```java { .api }
145
public interface ClusterDescriptor<T> extends AutoCloseable {
146
    ClusterClientProvider<T> retrieve(T clusterId) throws ClusterRetrieveException;
147
    ClusterClientProvider<T> deploySessionCluster(ClusterSpecification clusterSpecification) 
148
        throws ClusterDeploymentException;
149
}
150

151
public interface ClusterClientFactory<T> {
152
    boolean isCompatibleWith(Configuration configuration);
153
    ClusterDescriptor<T> createClusterDescriptor(Configuration configuration);
154
}
155
```
156

157
[Deployment Management](./deployment-management.md)
158

159
### Application Mode Deployment
160

161
Specialized deployment mode for long-running applications with dedicated cluster resources and lifecycle management.
162

163
```java { .api }
164
public interface ApplicationRunner {
165
    CompletableFuture<Void> run(
166
        DispatcherGateway dispatcherGateway,
167
        ScheduledExecutor scheduledExecutor,
168
        ApplicationConfiguration applicationConfiguration
169
    );
170
}
171

172
public class ApplicationConfiguration {
173
    public static Builder fromConfiguration(Configuration configuration);
174
    public String[] getProgramArguments();
175
    public String[] getApplicationArgs();
176
}
177
```
178

179
[Application Mode](./application-mode.md)
180

181
### Artifact Management
182

183
Flexible system for fetching and managing job artifacts from various sources including local files, HTTP endpoints, and distributed file systems.
184

185
```java { .api }
186
public interface ArtifactFetcher {
187
    CompletableFuture<File> fetch(
188
        URI uri,
189
        Configuration flinkConf,
190
        File targetDir,
191
        @Nullable String filename
192
    ) throws Exception;
193
}
194

195
public class ArtifactFetchManager {
196
    public static ArtifactFetchManager fromConfiguration(Configuration configuration);
197
    public CompletableFuture<File> fetchArtifact(URI uri, File targetDir) throws Exception;
198
}
199
```
200

201
[Artifact Management](./artifact-management.md)
202

203
## Types
204

205
```java { .api }
206
// Core types used across multiple capabilities
207
public class JobID extends AbstractID {
208
    public static JobID generate();
209
}
210

211
public enum JobStatus {
212
    CREATED, RUNNING, FAILING, FAILED, CANCELLING, CANCELED, FINISHED, RESTARTING, SUSPENDED, RECONCILING
213
}
214

215
public class JobStatusMessage {
216
    public JobID getJobId();
217
    public String getJobName();
218
    public JobStatus getJobState();
219
    public long getStartTime();
220
}
221

222
public class ClusterSpecification {
223
    public int getMasterMemoryMB();
224
    public int getTaskManagerMemoryMB();
225
    public int getNumberTaskManagers();
226
}
227

228
// Exception types
229
public class ProgramInvocationException extends FlinkException {
230
    public ProgramInvocationException(String message);
231
    public ProgramInvocationException(String message, Throwable cause);
232
}
233

234
public class ClusterDeploymentException extends FlinkException {
235
    public ClusterDeploymentException(String message);
236
    public ClusterDeploymentException(String message, Throwable cause);
237
}
238

239
public class ClusterRetrieveException extends FlinkException {
240
    public ClusterRetrieveException(String message);
241
    public ClusterRetrieveException(String message, Throwable cause);
242
}
243
```