Skip to content

Commit eea603b

Browse files
michaelpri10michaelpri10
authored
docs: Add samples and tests for ingestion from Kafka sources (#2315)
* docs: Add samples and tests for ingestion from Kafka sources * docs: Styles fixes for samples/tests
1 parent 5e80b57 commit eea603b

File tree

4 files changed

+355
-2
lines changed

4 files changed

+355
-2
lines changed

‎samples/snippets/src/main/java/pubsub/CreateTopicWithAwsMskIngestionExample.java‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
/*
2+
* Copyright 2025 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package pubsub;
18+
19+
// [START pubsub_create_topic_with_aws_msk_ingestion]
20+
21+
import com.google.cloud.pubsub.v1.TopicAdminClient;
22+
import com.google.cloud.pubsub.v1.TopicAdminSettings;
23+
import com.google.pubsub.v1.IngestionDataSourceSettings;
24+
import com.google.pubsub.v1.Topic;
25+
import com.google.pubsub.v1.TopicName;
26+
import java.io.IOException;
27+
28+
public class CreateTopicWithAwsMskIngestionExample {
29+
public static void main(String... args) throws Exception {
30+
// TODO(developer): Replace these variables before running the sample.
31+
String projectId = "your-project-id";
32+
String topicId = "your-topic-id";
33+
// AWS MSK ingestion settings.
34+
String clusterArn = "cluster-arn";
35+
String mskTopic = "msk-topic";
36+
String awsRoleArn = "aws-role-arn";
37+
String gcpServiceAccount = "gcp-service-account";
38+
39+
createTopicWithAwsMskIngestionExample(
40+
projectId, topicId, clusterArn, mskTopic, awsRoleArn, gcpServiceAccount);
41+
}
42+
43+
public static void createTopicWithAwsMskIngestionExample(
44+
String projectId,
45+
String topicId,
46+
String clusterArn,
47+
String mskTopic,
48+
String awsRoleArn,
49+
String gcpServiceAccount)
50+
throws IOException {
51+
try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
52+
TopicName topicName = TopicName.of(projectId, topicId);
53+
54+
IngestionDataSourceSettings.AwsMsk awsMsk =
55+
IngestionDataSourceSettings.AwsMsk.newBuilder()
56+
.setClusterArn(clusterArn)
57+
.setTopic(mskTopic)
58+
.setAwsRoleArn(awsRoleArn)
59+
.setGcpServiceAccount(gcpServiceAccount)
60+
.build();
61+
IngestionDataSourceSettings ingestionDataSourceSettings =
62+
IngestionDataSourceSettings.newBuilder().setAwsMsk(awsMsk).build();
63+
64+
Topic topic =
65+
topicAdminClient.createTopic(
66+
Topic.newBuilder()
67+
.setName(topicName.toString())
68+
.setIngestionDataSourceSettings(ingestionDataSourceSettings)
69+
.build());
70+
71+
System.out.println("Created topic with AWS MSK ingestion settings: " + topic.getAllFields());
72+
}
73+
}
74+
}
75+
// [END pubsub_create_topic_with_aws_msk_ingestion]

‎samples/snippets/src/main/java/pubsub/CreateTopicWithAzureEventHubsIngestionExample.java‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
/*
2+
* Copyright 2025 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package pubsub;
18+
19+
// [START pubsub_create_topic_with_azure_event_hubs_ingestion]
20+
21+
import com.google.cloud.pubsub.v1.TopicAdminClient;
22+
import com.google.cloud.pubsub.v1.TopicAdminSettings;
23+
import com.google.pubsub.v1.IngestionDataSourceSettings;
24+
import com.google.pubsub.v1.Topic;
25+
import com.google.pubsub.v1.TopicName;
26+
import java.io.IOException;
27+
28+
public class CreateTopicWithAzureEventHubsIngestionExample {
29+
public static void main(String... args) throws Exception {
30+
// TODO(developer): Replace these variables before running the sample.
31+
String projectId = "your-project-id";
32+
String topicId = "your-topic-id";
33+
// Azure Event Hubs ingestion settings.
34+
String resourceGroup = "resource-group";
35+
String namespace = "namespace";
36+
String eventHub = "event-hub";
37+
String clientId = "client-id";
38+
String tenantId = "tenant-id";
39+
String subscriptionId = "subscription-id";
40+
String gcpServiceAccount = "gcp-service-account";
41+
42+
createTopicWithAzureEventHubsIngestionExample(
43+
projectId,
44+
topicId,
45+
resourceGroup,
46+
namespace,
47+
eventHub,
48+
clientId,
49+
tenantId,
50+
subscriptionId,
51+
gcpServiceAccount);
52+
}
53+
54+
public static void createTopicWithAzureEventHubsIngestionExample(
55+
String projectId,
56+
String topicId,
57+
String resourceGroup,
58+
String namespace,
59+
String eventHub,
60+
String clientId,
61+
String tenantId,
62+
String subscriptionId,
63+
String gcpServiceAccount)
64+
throws IOException {
65+
try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
66+
TopicName topicName = TopicName.of(projectId, topicId);
67+
68+
IngestionDataSourceSettings.AzureEventHubs azureEventHubs =
69+
IngestionDataSourceSettings.AzureEventHubs.newBuilder()
70+
.setResourceGroup(resourceGroup)
71+
.setNamespace(namespace)
72+
.setEventHub(eventHub)
73+
.setClientId(clientId)
74+
.setTenantId(tenantId)
75+
.setSubscriptionId(subscriptionId)
76+
.setGcpServiceAccount(gcpServiceAccount)
77+
.build();
78+
IngestionDataSourceSettings ingestionDataSourceSettings =
79+
IngestionDataSourceSettings.newBuilder().setAzureEventHubs(azureEventHubs).build();
80+
81+
Topic topic =
82+
topicAdminClient.createTopic(
83+
Topic.newBuilder()
84+
.setName(topicName.toString())
85+
.setIngestionDataSourceSettings(ingestionDataSourceSettings)
86+
.build());
87+
88+
System.out.println(
89+
"Created topic with Azure Event Hubs ingestion settings: " + topic.getAllFields());
90+
}
91+
}
92+
}
93+
// [END pubsub_create_topic_with_azure_event_hubs_ingestion]

‎samples/snippets/src/main/java/pubsub/CreateTopicWithConfluentCloudIngestionExample.java‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/*
2+
* Copyright 2025 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package pubsub;
18+
19+
// [START pubsub_create_topic_with_confluent_cloud_ingestion]
20+
21+
import com.google.cloud.pubsub.v1.TopicAdminClient;
22+
import com.google.cloud.pubsub.v1.TopicAdminSettings;
23+
import com.google.pubsub.v1.IngestionDataSourceSettings;
24+
import com.google.pubsub.v1.Topic;
25+
import com.google.pubsub.v1.TopicName;
26+
import java.io.IOException;
27+
28+
public class CreateTopicWithConfluentCloudIngestionExample {
29+
public static void main(String... args) throws Exception {
30+
// TODO(developer): Replace these variables before running the sample.
31+
String projectId = "your-project-id";
32+
String topicId = "your-topic-id";
33+
// Confluent Cloud ingestion settings.
34+
String bootstrapServer = "bootstrap-server";
35+
String clusterId = "cluster-id";
36+
String confluentTopic = "confluent-topic";
37+
String identityPoolId = "identity-pool-id";
38+
String gcpServiceAccount = "gcp-service-account";
39+
40+
createTopicWithConfluentCloudIngestionExample(
41+
projectId,
42+
topicId,
43+
bootstrapServer,
44+
clusterId,
45+
confluentTopic,
46+
identityPoolId,
47+
gcpServiceAccount);
48+
}
49+
50+
public static void createTopicWithConfluentCloudIngestionExample(
51+
String projectId,
52+
String topicId,
53+
String bootstrapServer,
54+
String clusterId,
55+
String confluentTopic,
56+
String identityPoolId,
57+
String gcpServiceAccount)
58+
throws IOException {
59+
try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
60+
TopicName topicName = TopicName.of(projectId, topicId);
61+
62+
IngestionDataSourceSettings.ConfluentCloud confluentCloud =
63+
IngestionDataSourceSettings.ConfluentCloud.newBuilder()
64+
.setBootstrapServer(bootstrapServer)
65+
.setClusterId(clusterId)
66+
.setTopic(confluentTopic)
67+
.setIdentityPoolId(identityPoolId)
68+
.setGcpServiceAccount(gcpServiceAccount)
69+
.build();
70+
IngestionDataSourceSettings ingestionDataSourceSettings =
71+
IngestionDataSourceSettings.newBuilder().setConfluentCloud(confluentCloud).build();
72+
73+
Topic topic =
74+
topicAdminClient.createTopic(
75+
Topic.newBuilder()
76+
.setName(topicName.toString())
77+
.setIngestionDataSourceSettings(ingestionDataSourceSettings)
78+
.build());
79+
80+
System.out.println(
81+
"Created topic with Confluent Cloud ingestion settings: " + topic.getAllFields());
82+
}
83+
}
84+
}
85+
// [END pubsub_create_topic_with_confluent_cloud_ingestion]

‎samples/snippets/src/test/java/pubsub/AdminIT.java‎

Lines changed: 102 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,11 @@ public class AdminIT {
5555
private static final String kinesisIngestionTopicId = "kinesis-ingestion-topic-" + _suffix;
5656
private static final String cloudStorageIngestionTopicId =
5757
"cloud-storage-ingestion-topic-" + _suffix;
58+
private static final String awsMskIngestionTopicId = "aws-msk-ingestion-topic-" + _suffix;
59+
private static final String confluentCloudIngestionTopicId =
60+
"confluent-cloud-ingestion-topic-" + _suffix;
61+
private static final String azureEventHubsIngestionTopicId =
62+
"azure-event-hubs-ingestion-topic-" + _suffix;
5863
private static final String pullSubscriptionId = "iam-pull-subscription-" + _suffix;
5964
private static final String pushSubscriptionId = "iam-push-subscription-" + _suffix;
6065
private static final String orderedSubscriptionId = "iam-ordered-subscription-" + _suffix;
@@ -66,6 +71,9 @@ public class AdminIT {
6671
"java_samples_data_set" + _suffix.replace("-", "_");
6772
private static final String bigquerySubscriptionId = "iam-bigquery-subscription-" + _suffix;
6873
private static final String bigqueryTableId = "java_samples_table_" + _suffix;
74+
private static final String gcpServiceAccount =
75+
"fake-service-account@fake-gcp-project.iam.gserviceaccount.com";
76+
// AWS Kinesis ingestion settings.
6977
private static final String streamArn =
7078
"arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name";
7179
private static final String consumerArn =
@@ -75,20 +83,41 @@ public class AdminIT {
7583
"arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name/"
7684
+ "consumer/consumer-2:2222222222";
7785
private static final String awsRoleArn = "arn:aws:iam::111111111111:role/fake-role-name";
78-
private static final String gcpServiceAccount =
79-
"fake-service-account@fake-gcp-project.iam.gserviceaccount.com";
86+
// GCS ingestion settings.
8087
private static final String cloudStorageBucket = "pubsub-cloud-storage-bucket";
8188
private static final String cloudStorageInputFormat = "text";
8289
private static final String cloudStorageTextDelimiter = ",";
8390
private static final String cloudStorageMatchGlob = "**.txt";
8491
private static final String cloudStorageMinimumObjectCreateTime = "1970-01-01T00:00:01Z";
8592
private static final String cloudStorageMinimumObjectCreateTimeSeconds = "seconds: 1";
93+
// AWS MSK ingestion settings.
94+
String clusterArn =
95+
"arn:aws:kafka:us-east-1:111111111111:cluster/fake-cluster-name/11111111-1111-1";
96+
String mskTopic = "fake-msk-topic-name";
97+
// Confluent Cloud ingestion settings.
98+
String bootstrapServer = "fake-bootstrap-server-id.us-south1.gcp.confluent.cloud:9092";
99+
String clusterId = "fake-cluster-id";
100+
String confluentTopic = "fake-confluent-topic-name";
101+
String identityPoolId = "fake-pool-id";
102+
// Azure Event Hubs ingestion settings.
103+
String resourceGroup = "fake-resource-group";
104+
String namespace = "fake-namespace";
105+
String eventHub = "fake-event-hub";
106+
String clientId = "11111111-1111-1111-1111-111111111111";
107+
String tenantId = "22222222-2222-2222-2222-222222222222";
108+
String subscriptionId = "33333333-3333-3333-3333-333333333333";
86109

87110
private static final TopicName topicName = TopicName.of(projectId, topicId);
88111
private static final TopicName kinesisIngestionTopicName =
89112
TopicName.of(projectId, kinesisIngestionTopicId);
90113
private static final TopicName cloudStorageIngestionTopicName =
91114
TopicName.of(projectId, cloudStorageIngestionTopicId);
115+
private static final TopicName awsMskIngestionTopicName =
116+
TopicName.of(projectId, awsMskIngestionTopicId);
117+
private static final TopicName confluentCloudIngestionTopicName =
118+
TopicName.of(projectId, confluentCloudIngestionTopicId);
119+
private static final TopicName azureEventHubsIngestionTopicName =
120+
TopicName.of(projectId, azureEventHubsIngestionTopicId);
92121
private static final SubscriptionName pullSubscriptionName =
93122
SubscriptionName.of(projectId, pullSubscriptionId);
94123
private static final SubscriptionName pushSubscriptionName =
@@ -361,5 +390,76 @@ public void testAdmin() throws Exception {
361390
// Test delete Cloud Storage ingestion topic.
362391
DeleteTopicExample.deleteTopicExample(projectId, cloudStorageIngestionTopicId);
363392
assertThat(bout.toString()).contains("Deleted topic.");
393+
394+
bout.reset();
395+
// Test create topic with AWS MSK ingestion settings.
396+
CreateTopicWithAwsMskIngestionExample.createTopicWithAwsMskIngestionExample(
397+
projectId,
398+
awsMskIngestionTopicId,
399+
clusterArn,
400+
mskTopic,
401+
awsRoleArn,
402+
gcpServiceAccount);
403+
assertThat(bout.toString())
404+
.contains("google.pubsub.v1.Topic.name=" + awsMskIngestionTopicName.toString());
405+
assertThat(bout.toString()).contains(clusterArn);
406+
assertThat(bout.toString()).contains(mskTopic);
407+
assertThat(bout.toString()).contains(awsRoleArn);
408+
assertThat(bout.toString()).contains(gcpServiceAccount);
409+
410+
bout.reset();
411+
// Test delete AWS MSK ingestion topic.
412+
DeleteTopicExample.deleteTopicExample(projectId, awsMskIngestionTopicId);
413+
assertThat(bout.toString()).contains("Deleted topic.");
414+
415+
bout.reset();
416+
// Test create topic with Confluent Cloud ingestion settings.
417+
CreateTopicWithConfluentCloudIngestionExample.createTopicWithConfluentCloudIngestionExample(
418+
projectId,
419+
confluentCloudIngestionTopicId,
420+
bootstrapServer,
421+
clusterId,
422+
confluentTopic,
423+
identityPoolId,
424+
gcpServiceAccount);
425+
assertThat(bout.toString())
426+
.contains("google.pubsub.v1.Topic.name=" + confluentCloudIngestionTopicName.toString());
427+
assertThat(bout.toString()).contains(bootstrapServer);
428+
assertThat(bout.toString()).contains(clusterId);
429+
assertThat(bout.toString()).contains(confluentTopic);
430+
assertThat(bout.toString()).contains(identityPoolId);
431+
assertThat(bout.toString()).contains(gcpServiceAccount);
432+
433+
bout.reset();
434+
// Test delete Confluent Cloud ingestion topic.
435+
DeleteTopicExample.deleteTopicExample(projectId, confluentCloudIngestionTopicId);
436+
assertThat(bout.toString()).contains("Deleted topic.");
437+
438+
bout.reset();
439+
// Test create topic with Azure Event Hubs ingestion settings.
440+
CreateTopicWithAzureEventHubsIngestionExample.createTopicWithAzureEventHubsIngestionExample(
441+
projectId,
442+
azureEventHubsIngestionTopicId,
443+
resourceGroup,
444+
namespace,
445+
eventHub,
446+
clientId,
447+
tenantId,
448+
subscriptionId,
449+
gcpServiceAccount);
450+
assertThat(bout.toString()).contains(
451+
"google.pubsub.v1.Topic.name=" + azureEventHubsIngestionTopicName.toString());
452+
assertThat(bout.toString()).contains(resourceGroup);
453+
assertThat(bout.toString()).contains(namespace);
454+
assertThat(bout.toString()).contains(eventHub);
455+
assertThat(bout.toString()).contains(clientId);
456+
assertThat(bout.toString()).contains(tenantId);
457+
assertThat(bout.toString()).contains(subscriptionId);
458+
assertThat(bout.toString()).contains(gcpServiceAccount);
459+
460+
bout.reset();
461+
// Test delete Azure Event Hubs ingestion topic.
462+
DeleteTopicExample.deleteTopicExample(projectId, azureEventHubsIngestionTopicId);
463+
assertThat(bout.toString()).contains("Deleted topic.");
364464
}
365465
}

0 commit comments

Comments
 (0)