0
# Apache Flink Clients Library
1

2
The Apache Flink clients library (flink-clients_2.10) provides comprehensive APIs and utilities for programmatically interacting with Apache Flink clusters. It enables developers to submit batch and streaming jobs, monitor job execution, manage cluster resources, and handle job lifecycle operations from external applications.
3

4
## Package Information
5

6
- **Package Name**: org.apache.flink:flink-clients_2.10
7
- **Package Type**: Maven
8
- **Language**: Java (with Scala 2.10 compatibility)
9
- **Installation**: Add to Maven dependencies
10
  ```xml
11
  <dependency>
12
    <groupId>org.apache.flink</groupId>
13
    <artifactId>flink-clients_2.10</artifactId>
14
    <version>1.3.3</version>
15
  </dependency>
16
  ```
17

18
## Core Imports
19

20
```java
21
import org.apache.flink.client.CliFrontend;
22
import org.apache.flink.client.LocalExecutor;
23
import org.apache.flink.client.RemoteExecutor;
24
import org.apache.flink.client.program.ClusterClient;
25
import org.apache.flink.client.program.PackagedProgram;
26
import org.apache.flink.client.program.StandaloneClusterClient;
27
import org.apache.flink.client.program.ContextEnvironment;
28
```
29

30
## Basic Usage
31

32
```java
33
import org.apache.flink.client.program.PackagedProgram;
34
import org.apache.flink.client.program.StandaloneClusterClient;
35
import org.apache.flink.client.program.ProgramInvocationException;
36
import org.apache.flink.client.program.ProgramMissingJobException;
37
import org.apache.flink.client.JobExecutionException;
38
import org.apache.flink.configuration.Configuration;
39
import org.apache.flink.api.common.JobSubmissionResult;
40
import org.apache.flink.api.common.JobExecutionResult;
41
import org.apache.flink.api.common.JobID;
42
import java.io.File;
43

44
try {
45
    // Connect to a Flink cluster
46
    Configuration config = new Configuration();
47
    StandaloneClusterClient client = new StandaloneClusterClient(config);
48

49
    // Package and run a Flink program
50
    File jarFile = new File("path/to/your/flink-job.jar");
51
    PackagedProgram program = new PackagedProgram(jarFile, "arg1", "arg2");
52

53
    // Submit the job
54
    JobSubmissionResult result = client.run(program, 4); // 4 = parallelism
55
    JobID jobId = result.getJobID();
56

57
    // Monitor job execution
58
    JobExecutionResult executionResult = client.retrieveJob(jobId);
59
    System.out.println("Job finished with result: " + executionResult.getNetRuntime());
60

61
    // Clean up
62
    client.shutdown();
63
} catch (ProgramInvocationException | ProgramMissingJobException e) {
64
    System.err.println("Program execution failed: " + e.getMessage());
65
} catch (JobExecutionException e) {
66
    System.err.println("Job execution failed: " + e.getMessage());
67
} catch (Exception e) {
68
    System.err.println("Error: " + e.getMessage());
69
}
70
```
71

72
## Architecture
73

74
Apache Flink clients library is built around several key components:
75

76
- **Cluster Clients**: Connection and communication with Flink clusters (`ClusterClient`, `StandaloneClusterClient`)
77
- **Program Management**: JAR-based job packaging and execution (`PackagedProgram`, `JobWithJars`)
78
- **Execution Environments**: Context-aware execution environments for different deployment scenarios
79
- **Command Line Interface**: Full-featured CLI for job management and cluster operations
80
- **Deployment Descriptors**: Cluster deployment and connection abstractions
81

82
## Capabilities
83

84
### Cluster Connection and Management
85

86
Core functionality for connecting to and managing Flink clusters, including job submission, monitoring, and lifecycle operations.
87

88
```java { .api }
89
public abstract class ClusterClient {
90
    public ClusterClient(Configuration flinkConfig) throws Exception;
91
    public ClusterClient(Configuration flinkConfig, HighAvailabilityServices highAvailabilityServices);
92
    
93
    public JobSubmissionResult run(PackagedProgram prog, int parallelism) throws ProgramInvocationException, ProgramMissingJobException;
94
    public JobExecutionResult run(JobGraph jobGraph, ClassLoader classLoader) throws ProgramInvocationException;
95
    public JobSubmissionResult runDetached(JobGraph jobGraph, ClassLoader classLoader) throws ProgramInvocationException;
96
    
97
    public void cancel(JobID jobId) throws Exception;
98
    public void stop(JobID jobId) throws Exception;
99
    public Map<String, Object> getAccumulators(JobID jobID) throws Exception;
100
    public JobExecutionResult retrieveJob(JobID jobID) throws JobExecutionException;
101
    
102
    public void shutdown() throws Exception;
103
    public void setDetached(boolean isDetached);
104
    public boolean isDetached();
105
    
106
    public abstract void waitForClusterToBeReady();
107
    public abstract String getWebInterfaceURL();
108
    public abstract GetClusterStatusResponse getClusterStatus();
109
}
110

111
public class StandaloneClusterClient extends ClusterClient {
112
    public StandaloneClusterClient(Configuration config);
113
    public StandaloneClusterClient(Configuration config, HighAvailabilityServices highAvailabilityServices);
114
}
115
```
116

117
[Cluster Management](./cluster-management.md)
118

119
### Program Packaging and Execution
120

121
Utilities for packaging Flink programs as JAR files and managing their execution with dependencies and classpath handling.
122

123
```java { .api }
124
public class PackagedProgram {
125
    public static final String MANIFEST_ATTRIBUTE_ASSEMBLER_CLASS = "program-class";
126
    public static final String MANIFEST_ATTRIBUTE_MAIN_CLASS = "Main-Class";
127
    
128
    public PackagedProgram(File jarFile, String... args) throws ProgramInvocationException;
129
    public PackagedProgram(File jarFile, List<URL> classpaths, String... args) throws ProgramInvocationException;
130
    public PackagedProgram(File jarFile, String entryPointClassName, String... args) throws ProgramInvocationException;
131
    public PackagedProgram(File jarFile, List<URL> classpaths, String entryPointClassName, String... args) throws ProgramInvocationException;
132
    
133
    public String[] getArguments();
134
    public String getMainClassName();
135
    public JobWithJars getPlanWithJars() throws ProgramInvocationException;
136
    public JobWithJars getPlanWithoutJars() throws ProgramInvocationException;
137
    public ClassLoader getUserCodeClassLoader();
138
    public List<URL> getClasspaths();
139
    public String getDescription() throws ProgramInvocationException;
140
}
141

142
public class JobWithJars {
143
    public JobWithJars(Plan plan, List<URL> jarFiles, List<URL> classpaths);
144
    public Plan getPlan();
145
    public List<URL> getJarFiles();
146
    public List<URL> getClasspaths();
147
}
148
```
149

150
[Program Management](./program-management.md)
151

152
### Command Line Interface
153

154
Comprehensive command line interface for Flink cluster operations including job submission, monitoring, cancellation, and savepoint management.
155

156
```java { .api }
157
public class CliFrontend {
158
    public CliFrontend();
159
    public CliFrontend(String configDir);
160
    
161
    public int run(String[] args);
162
    public int info(String[] args);
163
    public int list(String[] args);
164
    public int cancel(String[] args);
165
    public int stop(String[] args);
166
    public int savepoint(String[] args);
167
    
168
    public static void main(String[] args);
169
}
170
```
171

172
[Command Line Interface](./cli.md)
173

174
### Local and Remote Execution
175

176
Executors for running Flink programs both locally for development and testing, and remotely on production clusters.
177

178
```java { .api }
179
public class LocalExecutor extends PlanExecutor {
180
    public LocalExecutor();
181
    public LocalExecutor(Configuration conf);
182
    
183
    public void start() throws Exception;
184
    public void stop() throws Exception;
185
    public JobExecutionResult executePlan(Plan plan) throws Exception;
186
    
187
    public static JobExecutionResult execute(Plan plan) throws Exception;
188
}
189

190
public class RemoteExecutor extends PlanExecutor {
191
    public RemoteExecutor(String hostname, int port);
192
    public RemoteExecutor(String hostname, int port, List<URL> jarFiles);
193
    public RemoteExecutor(InetSocketAddress inet, Configuration clientConfiguration, List<URL> jarFiles, List<URL> globalClasspaths);
194
    
195
    public JobExecutionResult executePlan(Plan plan) throws Exception;
196
}
197
```
198

199
[Execution Environments](./execution-environments.md)
200

201
### Programmatic Execution Context
202

203
Execution environments that provide programmatic APIs for submitting and managing Flink jobs within applications.
204

205
```java { .api }
206
public class ContextEnvironment extends ExecutionEnvironment {
207
    public ContextEnvironment(ClusterClient remoteConnection, List<URL> jarFiles, List<URL> classpaths, ClassLoader userCodeClassLoader, SavepointRestoreSettings savepointSettings);
208
    
209
    public JobExecutionResult execute(String jobName) throws Exception;
210
    public String getExecutionPlan() throws Exception;
211
    public ClusterClient getClient();
212
    
213
    public static void setAsContext(ContextEnvironmentFactory factory);
214
    public static void unsetContext();
215
}
216
```
217

218
[Execution Context](./execution-context.md)
219

220
## Types
221

222
```java { .api }
223
public class JobSubmissionResult {
224
    public JobID getJobID();
225
    public boolean isJobExecutionResult();
226
}
227

228
public class JobExecutionResult extends JobSubmissionResult {
229
    public long getNetRuntime();
230
    public Map<String, Object> getAllAccumulatorResults();
231
}
232

233
public class ProgramInvocationException extends Exception {
234
    public ProgramInvocationException(String message);
235
    public ProgramInvocationException(Throwable cause);
236
    public ProgramInvocationException(String message, Throwable cause);
237
}
238

239
public class ProgramMissingJobException extends ProgramInvocationException {
240
    public ProgramMissingJobException(String message);
241
}
242

243
public class JobExecutionException extends Exception {
244
    public JobExecutionException(JobID jobId, String message);
245
    public JobExecutionException(JobID jobId, String message, Throwable cause);
246
}
247

248
public class CompilerException extends Exception {
249
    public CompilerException();
250
    public CompilerException(String message);
251
    public CompilerException(Throwable cause);
252
    public CompilerException(String message, Throwable cause);
253
}
254

255
public class SavepointRestoreSettings {
256
    public static SavepointRestoreSettings none();
257
    public static SavepointRestoreSettings forPath(String savepointPath);
258
    public static SavepointRestoreSettings forPath(String savepointPath, boolean allowNonRestoredState);
259
    
260
    public boolean restoreSavepoint();
261
    public String getRestorePath();
262
    public boolean allowNonRestoredState();
263
}
264

265
public class GetClusterStatusResponse {
266
    public int numTaskManagersConnected();
267
    public int numSlotsTotal();
268
    public int numSlotsAvailable();
269
}
270

271
public class JobListeningContext {
272
    public JobExecutionResult getJobExecutionResult() throws Exception;
273
    public void cancel() throws Exception;
274
}
275
```