0
# Apache Flink GCP Pub/Sub Connector
1
2
Apache Flink connector for Google Cloud Pub/Sub enables consuming messages from and publishing messages to Google Pub/Sub topics with exactly-once processing guarantees. The connector provides both source and sink capabilities for real-time stream processing applications with automatic acknowledgment management through Flink's checkpointing mechanism.
3
4
## Package Information
5
6
- **Package Name**: flink-connector-gcp-pubsub_2.12
7
- **Package Type**: Maven
8
- **Group ID**: org.apache.flink
9
- **Language**: Java
10
- **Installation**:
11
```xml
12
<dependency>
13
<groupId>org.apache.flink</groupId>
14
<artifactId>flink-connector-gcp-pubsub_2.12</artifactId>
15
<version>1.14.6</version>
16
</dependency>
17
```
18
19
## Core Imports
20
21
```java
22
import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource;
23
import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSink;
24
import org.apache.flink.streaming.connectors.gcp.pubsub.common.PubSubDeserializationSchema;
25
import org.apache.flink.streaming.connectors.gcp.pubsub.common.PubSubSubscriberFactory;
26
import org.apache.flink.streaming.connectors.gcp.pubsub.common.PubSubSubscriber;
27
import org.apache.flink.api.common.serialization.DeserializationSchema;
28
import org.apache.flink.api.common.serialization.SerializationSchema;
29
import org.apache.flink.util.Collector;
30
```
31
32
For credentials management:
33
```java
34
import com.google.auth.Credentials;
35
import com.google.cloud.pubsub.v1.SubscriptionAdminSettings;
36
```
37
38
For Google Cloud Pub/Sub types:
39
```java
40
import com.google.pubsub.v1.PubsubMessage;
41
import com.google.pubsub.v1.ReceivedMessage;
42
import java.util.List;
43
import java.time.Duration;
44
```
45
46
For emulator testing:
47
```java
48
import org.apache.flink.streaming.connectors.gcp.pubsub.emulator.EmulatorCredentials;
49
import org.apache.flink.streaming.connectors.gcp.pubsub.emulator.EmulatorCredentialsProvider;
50
import org.apache.flink.streaming.connectors.gcp.pubsub.emulator.PubSubSubscriberFactoryForEmulator;
51
```
52
53
## Basic Usage
54
55
### Consuming Messages (Source)
56
57
```java
58
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
59
import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSource;
60
import org.apache.flink.api.common.serialization.SimpleStringSchema;
61
62
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
63
64
// Enable checkpointing (required for exactly-once guarantees)
65
env.enableCheckpointing(30000); // checkpoint every 30 seconds
66
67
// Create PubSubSource
68
PubSubSource<String> pubsubSource = PubSubSource.newBuilder()
69
.withDeserializationSchema(new SimpleStringSchema())
70
.withProjectName("my-gcp-project")
71
.withSubscriptionName("my-subscription")
72
.build();
73
74
// Add source to stream
75
DataStream<String> stream = env.addSource(pubsubSource);
76
stream.print();
77
78
env.execute("PubSub Consumer");
79
```
80
81
### Publishing Messages (Sink)
82
83
```java
84
import org.apache.flink.streaming.connectors.gcp.pubsub.PubSubSink;
85
import org.apache.flink.api.common.serialization.SimpleStringSchema;
86
87
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
88
89
// Create data stream
90
DataStream<String> inputStream = env.fromElements("Hello", "World", "PubSub");
91
92
// Create PubSubSink
93
PubSubSink<String> pubsubSink = PubSubSink.newBuilder()
94
.withSerializationSchema(new SimpleStringSchema())
95
.withProjectName("my-gcp-project")
96
.withTopicName("my-topic")
97
.build();
98
99
// Add sink to stream
100
inputStream.addSink(pubsubSink);
101
102
env.execute("PubSub Producer");
103
```
104
105
## Architecture
106
107
The Apache Flink GCP Pub/Sub connector is built around several key components:
108
109
- **PubSubSource**: Source function for consuming messages with exactly-once processing guarantees
110
- **PubSubSink**: Sink function for publishing messages with at-least-once delivery semantics
111
- **Builder Pattern**: Fluent API for configuring both sources and sinks with optional parameters
112
- **Checkpointing Integration**: Automatic message acknowledgment tied to Flink checkpoint completion
113
- **Rate Limiting**: Configurable message consumption rate limits per parallel subtask
114
- **Emulator Support**: Built-in support for Google Pub/Sub emulator for testing scenarios
115
116
The connector ensures data consistency through Flink's distributed checkpointing mechanism, where Pub/Sub messages are only acknowledged after successful checkpoint completion, preventing message loss during failure scenarios.
117
118
## Capabilities
119
120
### PubSub Message Source
121
122
Source functionality for consuming messages from Google Cloud Pub/Sub subscriptions with exactly-once processing guarantees and configurable rate limiting.
123
124
```java { .api }
125
public static DeserializationSchemaBuilder newBuilder();
126
127
public static class PubSubSourceBuilder<OUT> {
128
public PubSubSourceBuilder<OUT> withCredentials(Credentials credentials);
129
public PubSubSourceBuilder<OUT> withPubSubSubscriberFactory(PubSubSubscriberFactory factory);
130
public PubSubSourceBuilder<OUT> withPubSubSubscriberFactory(int maxMessagesPerPull, Duration perRequestTimeout, int retries);
131
public PubSubSourceBuilder<OUT> withMessageRateLimit(int messagePerSecondRateLimit);
132
public PubSubSource<OUT> build();
133
}
134
```
135
136
[PubSub Source](./pubsub-source.md)
137
138
### PubSub Message Sink
139
140
Sink functionality for publishing messages to Google Cloud Pub/Sub topics with reliable delivery and checkpoint synchronization.
141
142
```java { .api }
143
public static SerializationSchemaBuilder newBuilder();
144
145
public static class PubSubSinkBuilder<IN> {
146
public PubSubSinkBuilder<IN> withCredentials(Credentials credentials);
147
public PubSubSinkBuilder<IN> withHostAndPortForEmulator(String hostAndPort);
148
public PubSubSink<IN> build();
149
}
150
```
151
152
[PubSub Sink](./pubsub-sink.md)
153
154
### Testing Support
155
156
Emulator support for local development and testing scenarios without requiring actual Google Cloud Pub/Sub infrastructure.
157
158
```java { .api }
159
public final class EmulatorCredentials extends OAuth2Credentials {
160
public static EmulatorCredentials getInstance();
161
}
162
163
public final class EmulatorCredentialsProvider implements CredentialsProvider {
164
public static EmulatorCredentialsProvider create();
165
}
166
```
167
168
[Emulator Testing](./emulator-testing.md)
169
170
## Core Interfaces
171
172
### PubSubDeserializationSchema
173
174
Interface for custom deserialization with access to full PubSub message metadata.
175
176
```java { .api }
177
public interface PubSubDeserializationSchema<T> extends Serializable, ResultTypeQueryable<T> {
178
default void open(DeserializationSchema.InitializationContext context) throws Exception;
179
boolean isEndOfStream(T nextElement);
180
T deserialize(PubsubMessage message) throws Exception;
181
default void deserialize(PubsubMessage message, Collector<T> out) throws Exception;
182
}
183
```
184
185
### PubSubSubscriberFactory
186
187
Factory interface for creating custom Pub/Sub subscribers with specialized configurations.
188
189
```java { .api }
190
public interface PubSubSubscriberFactory extends Serializable {
191
PubSubSubscriber getSubscriber(Credentials credentials) throws IOException;
192
}
193
```
194
195
### Checkpointing Support Classes
196
197
Core classes for managing checkpoint-based message acknowledgment.
198
199
```java { .api }
200
public interface Acknowledger<AcknowledgeId> {
201
void acknowledge(List<AcknowledgeId> ids);
202
}
203
204
public interface PubSubSubscriber extends Acknowledger<String> {
205
List<ReceivedMessage> pull();
206
void close() throws Exception;
207
}
208
209
public class AcknowledgeIdsForCheckpoint<AcknowledgeId> implements Serializable {
210
AcknowledgeIdsForCheckpoint(long checkpointId, List<AcknowledgeId> acknowledgeIds);
211
public long getCheckpointId();
212
public void setCheckpointId(long checkpointId);
213
public List<AcknowledgeId> getAcknowledgeIds();
214
public void setAcknowledgeIds(List<AcknowledgeId> acknowledgeIds);
215
}
216
217
public class AcknowledgeOnCheckpoint<ACKID extends Serializable>
218
implements ListCheckpointed<AcknowledgeIdsForCheckpoint<ACKID>>, CheckpointListener {
219
public AcknowledgeOnCheckpoint(Acknowledger<ACKID> acknowledger);
220
public void addAcknowledgeId(ACKID acknowledgeId);
221
public void acknowledgeIdsUpToCheckpoint(long checkpointId);
222
}
223
```
224
225
## Important Notes
226
227
- **Checkpointing Required**: PubSubSource requires Flink checkpointing to be enabled for exactly-once guarantees
228
- **Parallel Processing**: Both source and sink support parallel execution across multiple Flink subtasks
229
- **Rate Limiting**: Default rate limit is 100,000 messages per second per parallel source instance
230
- **Credentials**: If not explicitly provided, credentials are automatically loaded from the environment
231
- **Error Handling**: Failed message publishing in sink will cause job failure during checkpointing