Skip to content
This repository was archived by the owner on Mar 17, 2026. It is now read-only.

Commit c1bed9e

Browse files
michaelpri10michaelpri10
authored
docs: Add ingestion from Kafka samples and tests (#2009)
* docs: Add ingestion from Kafka samples and tests * chore: Fix formatting in samples
1 parent 506301b commit c1bed9e

7 files changed

Lines changed: 624 additions & 0 deletions

‎samples/createTopicWithAwsMskIngestion.js‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
// Copyright 2025 Google LLC
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// https://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// This is a generated sample, using the typeless sample bot. Please
16+
// look for the source TypeScript sample (.ts) for modifications.
17+
'use strict';
18+
19+
/**
20+
* This sample demonstrates how to perform basic operations on topics with
21+
* the Google Cloud Pub/Sub API.
22+
*
23+
* For more information, see the README.md under /pubsub and the documentation
24+
* at https://cloud.google.com/pubsub/docs.
25+
*/
26+
27+
// sample-metadata:
28+
// title: Create Topic With AWS MSK Ingestion
29+
// description: Creates a new topic, with AWS MSK ingestion enabled.
30+
// usage: node createTopicWithAwsMskIngestion.js <topic-name> <cluster-arn> <msk-topic> <role-arn> <gcp-service-account>
31+
32+
// [START pubsub_create_topic_with_aws_msk_ingestion]
33+
/**
34+
* TODO(developer): Uncomment these variables before running the sample.
35+
*/
36+
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
37+
// const clusterArn = 'arn:aws:kafka:...';
38+
// const mskTopic = 'YOUR_MSK_TOPIC';
39+
// const roleArn = 'arn:aws:iam:...';
40+
// const gcpServiceAccount = 'ingestion-account@...';
41+
42+
// Imports the Google Cloud client library
43+
const {PubSub} = require('@google-cloud/pubsub');
44+
45+
// Creates a client; cache this for further use
46+
const pubSubClient = new PubSub();
47+
48+
async function createTopicWithAwsMskIngestion(
49+
topicNameOrId,
50+
clusterArn,
51+
mskTopic,
52+
awsRoleArn,
53+
gcpServiceAccount
54+
) {
55+
// Creates a new topic with AWS MSK ingestion.
56+
await pubSubClient.createTopic({
57+
name: topicNameOrId,
58+
ingestionDataSourceSettings: {
59+
awsMsk: {
60+
clusterArn,
61+
topic: mskTopic,
62+
awsRoleArn,
63+
gcpServiceAccount,
64+
},
65+
},
66+
});
67+
console.log(`Topic ${topicNameOrId} created with AWS MSK ingestion.`);
68+
}
69+
// [END pubsub_create_topic_with_aws_msk_ingestion]
70+
71+
function main(
72+
topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID',
73+
clusterArn = 'arn:aws:kafka:...',
74+
mskTopic = 'YOUR_MSK_TOPIC',
75+
roleArn = 'arn:aws:iam:...',
76+
gcpServiceAccount = 'ingestion-account@...'
77+
) {
78+
createTopicWithAwsMskIngestion(
79+
topicNameOrId,
80+
clusterArn,
81+
mskTopic,
82+
roleArn,
83+
gcpServiceAccount
84+
).catch(err => {
85+
console.error(err.message);
86+
process.exitCode = 1;
87+
});
88+
}
89+
90+
main(...process.argv.slice(2));

‎samples/createTopicWithAzureEventHubsIngestion.js‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
// Copyright 2025 Google LLC
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// https://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// This is a generated sample, using the typeless sample bot. Please
16+
// look for the source TypeScript sample (.ts) for modifications.
17+
'use strict';
18+
19+
/**
20+
* This sample demonstrates how to perform basic operations on topics with
21+
* the Google Cloud Pub/Sub API.
22+
*
23+
* For more information, see the README.md under /pubsub and the documentation
24+
* at https://cloud.google.com/pubsub/docs.
25+
*/
26+
27+
// sample-metadata:
28+
// title: Create Topic With Azure Event Hubs Ingestion
29+
// description: Creates a new topic, with Azure Event Hubs ingestion enabled.
30+
// usage: node createTopicWithAzureEventHubsIngestion.js <topic-name> <cluster-arn> <msk-topic> <role-arn> <gcp-service-account>
31+
32+
// [START pubsub_create_topic_with_azure_event_hubs_ingestion]
33+
/**
34+
* TODO(developer): Uncomment these variables before running the sample.
35+
*/
36+
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
37+
// const resourceGroup = 'YOUR_RESOURCE_GROUP';
38+
// const namespace = 'YOUR_NAMESPACE';
39+
// const eventHub = 'YOUR_EVENT_HUB';
40+
// const clientId = 'YOUR_CLIENT_ID';
41+
// const tenantId = 'YOUR_TENANT_ID';
42+
// const subscriptionId = 'YOUR_SUBSCRIPTION_ID';
43+
// const gcpServiceAccount = 'ingestion-account@...';
44+
45+
// Imports the Google Cloud client library
46+
const {PubSub} = require('@google-cloud/pubsub');
47+
48+
// Creates a client; cache this for further use
49+
const pubSubClient = new PubSub();
50+
51+
async function createTopicWithAzureEventHubsIngestion(
52+
topicNameOrId,
53+
resourceGroup,
54+
namespace,
55+
eventHub,
56+
clientId,
57+
tenantId,
58+
subscriptionId,
59+
gcpServiceAccount
60+
) {
61+
// Creates a new topic with Azure Event Hubs ingestion.
62+
await pubSubClient.createTopic({
63+
name: topicNameOrId,
64+
ingestionDataSourceSettings: {
65+
azureEventHubs: {
66+
resourceGroup,
67+
namespace,
68+
eventHub,
69+
clientId,
70+
tenantId,
71+
subscriptionId,
72+
gcpServiceAccount,
73+
},
74+
},
75+
});
76+
console.log(
77+
`Topic ${topicNameOrId} created with Azure Event Hubs ingestion.`
78+
);
79+
}
80+
// [END pubsub_create_topic_with_azure_event_hubs_ingestion]
81+
82+
function main(
83+
topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID',
84+
resourceGroup = 'YOUR_RESOURCE_GROUP',
85+
namespace = 'YOUR_NAMESPACE',
86+
eventHub = 'YOUR_EVENT_HUB',
87+
clientId = 'YOUR_CLIENT_ID',
88+
tenantId = 'YOUR_TENANT_ID',
89+
subscriptionId = 'YOUR_SUBSCRIPTION_ID',
90+
gcpServiceAccount = 'ingestion-account@...'
91+
) {
92+
createTopicWithAzureEventHubsIngestion(
93+
topicNameOrId,
94+
resourceGroup,
95+
namespace,
96+
eventHub,
97+
clientId,
98+
tenantId,
99+
subscriptionId,
100+
gcpServiceAccount
101+
).catch(err => {
102+
console.error(err.message);
103+
process.exitCode = 1;
104+
});
105+
}
106+
107+
main(...process.argv.slice(2));

‎samples/createTopicWithConfluentCloudIngestion.js‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
// Copyright 2025 Google LLC
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// https://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// This is a generated sample, using the typeless sample bot. Please
16+
// look for the source TypeScript sample (.ts) for modifications.
17+
'use strict';
18+
19+
/**
20+
* This sample demonstrates how to perform basic operations on topics with
21+
* the Google Cloud Pub/Sub API.
22+
*
23+
* For more information, see the README.md under /pubsub and the documentation
24+
* at https://cloud.google.com/pubsub/docs.
25+
*/
26+
27+
// sample-metadata:
28+
// title: Create Topic With Confluent Cloud Ingestion
29+
// description: Creates a new topic, with Confluent Cloud ingestion enabled.
30+
// usage: node createTopicWithConfluentCloudIngestion.js <topic-name> <bootstrap-server> <cluster-id> <confluent-topic> <identity-pool-id> <gcp-service-account>
31+
32+
// [START pubsub_create_topic_with_confluent_cloud_ingestion]
33+
/**
34+
* TODO(developer): Uncomment these variables before running the sample.
35+
*/
36+
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
37+
// const bootstrapServer = 'url:port';
38+
// const clusterId = 'YOUR_CLUSTER_ID';
39+
// const confluentTopic = 'YOUR_CONFLUENT_TOPIC';
40+
// const identityPoolId = 'pool-ID';
41+
// const gcpServiceAccount = 'ingestion-account@...';
42+
43+
// Imports the Google Cloud client library
44+
const {PubSub} = require('@google-cloud/pubsub');
45+
46+
// Creates a client; cache this for further use
47+
const pubSubClient = new PubSub();
48+
49+
async function createTopicWithConfluentCloudIngestion(
50+
topicNameOrId,
51+
bootstrapServer,
52+
clusterId,
53+
confluentTopic,
54+
identityPoolId,
55+
gcpServiceAccount
56+
) {
57+
// Creates a new topic with Confluent Cloud ingestion.
58+
await pubSubClient.createTopic({
59+
name: topicNameOrId,
60+
ingestionDataSourceSettings: {
61+
confluentCloud: {
62+
bootstrapServer,
63+
clusterId,
64+
topic: confluentTopic,
65+
identityPoolId,
66+
gcpServiceAccount,
67+
},
68+
},
69+
});
70+
console.log(`Topic ${topicNameOrId} created with Confluent Cloud ingestion.`);
71+
}
72+
// [END pubsub_create_topic_with_confluent_cloud_ingestion]
73+
74+
function main(
75+
topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID',
76+
bootstrapServer = 'url:port',
77+
clusterId = 'YOUR_CLUSTER_ID',
78+
confluentTopic = 'YOUR_CONFLUENT_TOPIC',
79+
identityPoolId = 'pool-ID',
80+
gcpServiceAccount = 'ingestion-account@...'
81+
) {
82+
createTopicWithConfluentCloudIngestion(
83+
topicNameOrId,
84+
bootstrapServer,
85+
clusterId,
86+
confluentTopic,
87+
identityPoolId,
88+
gcpServiceAccount
89+
).catch(err => {
90+
console.error(err.message);
91+
process.exitCode = 1;
92+
});
93+
}
94+
95+
main(...process.argv.slice(2));

‎samples/system-test/topics.test.ts‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,58 @@ describe('topics', () => {
154154
}
155155
});
156156

157+
it('should create a topic with aws msk ingestion', async () => {
158+
const testId = 'create-aws-msk-ingestion';
159+
const name = topicName(testId);
160+
const clusterArn = 'arn:aws:kafka:us-east-1:111111111111:cluster/fake-cluster-name/11111111-1111-1';
161+
const mskTopic = 'fake-msk-topic';
162+
const roleArn = 'arn:aws:iam::111111111111:role/fake-role-name';
163+
const gcpServiceAccount = 'fake-service-account@fake-gcp-project.iam.gserviceaccount.com'
164+
165+
const output = execSync(
166+
`${commandFor('createTopicWithAwsMskIngestion')} ${name} ${clusterArn} ${mskTopic} ${roleArn} ${gcpServiceAccount}`);
167+
assert.include(output, `Topic ${name} created with AWS MSK ingestion.`)
168+
const [topics] = await pubsub.getTopics();
169+
const exists = topics.some(t => t.name === fullTopicName(name));
170+
assert.ok(exists, 'Topic was created');
171+
});
172+
173+
it('should create a topic with confluent cloud ingestion', async () => {
174+
const testId = 'create-confluent-cloud-ingestion';
175+
const name = topicName(testId);
176+
const bootstrapServer = 'fake-bootstrap-server-id.us-south1.gcp.confluent.cloud:9092';
177+
const clusterId = 'fake-cluster-id';
178+
const confluentTopic = 'fake-confluent-topic';
179+
const identityPoolId = 'fake-pool-id';
180+
const gcpServiceAccount = 'fake-service-account@fake-gcp-project.iam.gserviceaccount.com'
181+
182+
const output = execSync(
183+
`${commandFor('createTopicWithConfluentCloudIngestion')} ${name} ${bootstrapServer} ${clusterId} ${confluentTopic} ${identityPoolId} ${gcpServiceAccount}`);
184+
assert.include(output, `Topic ${name} created with Confluent Cloud ingestion.`)
185+
const [topics] = await pubsub.getTopics();
186+
const exists = topics.some(t => t.name === fullTopicName(name));
187+
assert.ok(exists, 'Topic was created');
188+
});
189+
190+
it('should create a topic with azure event hubs ingestion', async () => {
191+
const testId = 'create-azure-event-hubs-ingestion';
192+
const name = topicName(testId);
193+
const resourceGroup = 'fake-resource-group';
194+
const namespace = 'fake-namespace';
195+
const eventHub = 'fake-event-hub';
196+
const clientId = 'fake-client-id';
197+
const tenantId = 'fake-tenant-id';
198+
const subscriptionId = 'fake-subscription-id';
199+
const gcpServiceAccount = 'fake-service-account@fake-gcp-project.iam.gserviceaccount.com'
200+
201+
const output = execSync(
202+
`${commandFor('createTopicWithAzureEventHubsIngestion')} ${name} ${resourceGroup} ${namespace} ${eventHub} ${clientId} ${tenantId} ${subscriptionId} ${gcpServiceAccount}`);
203+
assert.include(output, `Topic ${name} created with Azure Event Hubs ingestion.`)
204+
const [topics] = await pubsub.getTopics();
205+
const exists = topics.some(t => t.name === fullTopicName(name));
206+
assert.ok(exists, 'Topic was created');
207+
});
208+
157209
it('should update a topic with kinesis integration', async () => {
158210
const pair = await createPair('update-kinesis');
159211
const output = execSync(

0 commit comments

Comments
 (0)