Skip to content

Commit 5b5c14b

Browse files
michaelpri10gcf-owl-bot[bot]
michaelpri10
and
gcf-owl-bot[bot]
authored
docs: Add Kinesis ingestion samples (#1947)
* docs: Adding samples for creating and updating Kinesis ingestion topics * style: Format AdmitIT * style: Re-format AdmitIT * docs: Fix ingestion topic deletion in test * docs: Add test for updating existing ingestion settings and verify the actual results of topic creation and updating * docs: Use response to verify ingestion settings update * style: fix formatting in tests * docs: Update AdminIT test to check for correct topic name * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
1 parent d5fe708 commit 5b5c14b

File tree

4 files changed

+212
-0
lines changed

4 files changed

+212
-0
lines changed

‎README.md

+2
Original file line numberDiff line numberDiff line change
@@ -257,6 +257,7 @@ Samples are in the [`samples/`](https://github.com/googleapis/java-pubsub/tree/m
257257
| Create Subscription With Filtering | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/CreateSubscriptionWithFiltering.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/CreateSubscriptionWithFiltering.java) |
258258
| Create Subscription With Ordering | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/CreateSubscriptionWithOrdering.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/CreateSubscriptionWithOrdering.java) |
259259
| Create Topic Example | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/CreateTopicExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/CreateTopicExample.java) |
260+
| Create Topic With Kinesis Ingestion Example | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/CreateTopicWithKinesisIngestionExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/CreateTopicWithKinesisIngestionExample.java) |
260261
| Create Topic With Schema Example | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/CreateTopicWithSchemaExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/CreateTopicWithSchemaExample.java) |
261262
| Create Topic With Schema Revisions Example | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/CreateTopicWithSchemaRevisionsExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/CreateTopicWithSchemaRevisionsExample.java) |
262263
| Create Unwrapped Push Subscription Example | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/CreateUnwrappedPushSubscriptionExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/CreateUnwrappedPushSubscriptionExample.java) |
@@ -307,6 +308,7 @@ Samples are in the [`samples/`](https://github.com/googleapis/java-pubsub/tree/m
307308
| Update Dead Letter Policy Example | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/UpdateDeadLetterPolicyExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/UpdateDeadLetterPolicyExample.java) |
308309
| Update Push Configuration Example | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/UpdatePushConfigurationExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/UpdatePushConfigurationExample.java) |
309310
| Update Topic Schema Example | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/UpdateTopicSchemaExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/UpdateTopicSchemaExample.java) |
311+
| Update Topic Type Example | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/UpdateTopicTypeExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/UpdateTopicTypeExample.java) |
310312
| Use Pub Sub Emulator Example | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/pubsub/UsePubSubEmulatorExample.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/pubsub/UsePubSubEmulatorExample.java) |
311313
| State | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/utilities/State.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/utilities/State.java) |
312314
| State Proto | [source code](https://github.com/googleapis/java-pubsub/blob/main/samples/snippets/src/main/java/utilities/StateProto.java) | [![Open in Cloud Shell][shell_img]](https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/googleapis/java-pubsub&page=editor&open_in_editor=samples/snippets/src/main/java/utilities/StateProto.java) |

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

+74
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
/*
2+
* Copyright 2024 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_kinesis_ingestion]
20+
21+
import com.google.cloud.pubsub.v1.TopicAdminClient;
22+
import com.google.pubsub.v1.IngestionDataSourceSettings;
23+
import com.google.pubsub.v1.Topic;
24+
import com.google.pubsub.v1.TopicName;
25+
import java.io.IOException;
26+
27+
public class CreateTopicWithKinesisIngestionExample {
28+
public static void main(String... args) throws Exception {
29+
// TODO(developer): Replace these variables before running the sample.
30+
String projectId = "your-project-id";
31+
String topicId = "your-topic-id";
32+
// Kinesis ingestion settings.
33+
String streamArn = "stream-arn";
34+
String consumerArn = "consumer-arn";
35+
String awsRoleArn = "aws-role-arn";
36+
String gcpServiceAccount = "gcp-service-account";
37+
38+
createTopicWithKinesisIngestionExample(
39+
projectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
40+
}
41+
42+
public static void createTopicWithKinesisIngestionExample(
43+
String projectId,
44+
String topicId,
45+
String streamArn,
46+
String consumerArn,
47+
String awsRoleArn,
48+
String gcpServiceAccount)
49+
throws IOException {
50+
try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
51+
TopicName topicName = TopicName.of(projectId, topicId);
52+
53+
IngestionDataSourceSettings.AwsKinesis awsKinesis =
54+
IngestionDataSourceSettings.AwsKinesis.newBuilder()
55+
.setStreamArn(streamArn)
56+
.setConsumerArn(consumerArn)
57+
.setAwsRoleArn(awsRoleArn)
58+
.setGcpServiceAccount(gcpServiceAccount)
59+
.build();
60+
IngestionDataSourceSettings ingestionDataSourceSettings =
61+
IngestionDataSourceSettings.newBuilder().setAwsKinesis(awsKinesis).build();
62+
63+
Topic topic =
64+
topicAdminClient.createTopic(
65+
Topic.newBuilder()
66+
.setName(topicName.toString())
67+
.setIngestionDataSourceSettings(ingestionDataSourceSettings)
68+
.build());
69+
70+
System.out.println("Created topic with Kinesis ingestion settings: " + topic.getAllFields());
71+
}
72+
}
73+
}
74+
// [END pubsub_create_topic_with_kinesis_ingestion]

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

+86
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
/*
2+
* Copyright 2024 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_update_topic_type]
20+
21+
import com.google.cloud.pubsub.v1.TopicAdminClient;
22+
import com.google.protobuf.FieldMask;
23+
import com.google.pubsub.v1.IngestionDataSourceSettings;
24+
import com.google.pubsub.v1.Topic;
25+
import com.google.pubsub.v1.TopicName;
26+
import com.google.pubsub.v1.UpdateTopicRequest;
27+
import java.io.IOException;
28+
29+
public class UpdateTopicTypeExample {
30+
public static void main(String... args) throws Exception {
31+
// TODO(developer): Replace these variables before running the sample.
32+
String projectId = "your-project-id";
33+
String topicId = "your-topic-id";
34+
// Kinesis ingestion settings.
35+
String streamArn = "stream-arn";
36+
String consumerArn = "consumer-arn";
37+
String awsRoleArn = "aws-role-arn";
38+
String gcpServiceAccount = "gcp-service-account";
39+
40+
UpdateTopicTypeExample.updateTopicTypeExample(
41+
projectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
42+
}
43+
44+
public static void updateTopicTypeExample(
45+
String projectId,
46+
String topicId,
47+
String streamArn,
48+
String consumerArn,
49+
String awsRoleArn,
50+
String gcpServiceAccount)
51+
throws IOException {
52+
try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
53+
TopicName topicName = TopicName.of(projectId, topicId);
54+
55+
IngestionDataSourceSettings.AwsKinesis awsKinesis =
56+
IngestionDataSourceSettings.AwsKinesis.newBuilder()
57+
.setStreamArn(streamArn)
58+
.setConsumerArn(consumerArn)
59+
.setAwsRoleArn(awsRoleArn)
60+
.setGcpServiceAccount(gcpServiceAccount)
61+
.build();
62+
IngestionDataSourceSettings ingestionDataSourceSettings =
63+
IngestionDataSourceSettings.newBuilder().setAwsKinesis(awsKinesis).build();
64+
65+
// Construct the topic with Kinesis ingestion settings.
66+
Topic topic =
67+
Topic.newBuilder()
68+
.setName(topicName.toString())
69+
.setIngestionDataSourceSettings(ingestionDataSourceSettings)
70+
.build();
71+
72+
// Construct a field mask to indicate which field to update in the topic.
73+
FieldMask updateMask =
74+
FieldMask.newBuilder().addPaths("ingestion_data_source_settings").build();
75+
76+
UpdateTopicRequest request =
77+
UpdateTopicRequest.newBuilder().setTopic(topic).setUpdateMask(updateMask).build();
78+
79+
Topic response = topicAdminClient.updateTopic(request);
80+
81+
System.out.println(
82+
"Updated topic with Kinesis ingestion settings: " + response.getAllFields());
83+
}
84+
}
85+
}
86+
// [END pubsub_update_topic_type]

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

+50
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@ public class AdminIT {
5252
private static final String projectId = System.getenv("GOOGLE_CLOUD_PROJECT");
5353
private static final String _suffix = UUID.randomUUID().toString();
5454
private static final String topicId = "iam-topic-" + _suffix;
55+
private static final String ingestionTopicId = "ingestion-topic-" + _suffix;
5556
private static final String pullSubscriptionId = "iam-pull-subscription-" + _suffix;
5657
private static final String pushSubscriptionId = "iam-push-subscription-" + _suffix;
5758
private static final String orderedSubscriptionId = "iam-ordered-subscription-" + _suffix;
@@ -63,8 +64,20 @@ public class AdminIT {
6364
"java_samples_data_set" + _suffix.replace("-", "_");
6465
private static final String bigquerySubscriptionId = "iam-bigquery-subscription-" + _suffix;
6566
private static final String bigqueryTableId = "java_samples_table_" + _suffix;
67+
private static final String streamArn =
68+
"arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name";
69+
private static final String consumerArn =
70+
"arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name/"
71+
+ "consumer/consumer-1:1111111111";
72+
private static final String consumerArn2 =
73+
"arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name/"
74+
+ "consumer/consumer-2:2222222222";
75+
private static final String awsRoleArn = "arn:aws:iam::111111111111:role/fake-role-name";
76+
private static final String gcpServiceAccount =
77+
"[email protected]";
6678

6779
private static final TopicName topicName = TopicName.of(projectId, topicId);
80+
private static final TopicName ingestionTopicName = TopicName.of(projectId, ingestionTopicId);
6881
private static final SubscriptionName pullSubscriptionName =
6982
SubscriptionName.of(projectId, pullSubscriptionId);
7083
private static final SubscriptionName pushSubscriptionName =
@@ -273,9 +286,46 @@ public void testAdmin() throws Exception {
273286
DeleteSubscriptionExample.deleteSubscriptionExample(projectId, bigquerySubscriptionId);
274287
assertThat(bout.toString()).contains("Deleted subscription.");
275288

289+
bout.reset();
290+
// Update topic type to Kinesis ingestion.
291+
UpdateTopicTypeExample.updateTopicTypeExample(
292+
projectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
293+
assertThat(bout.toString()).contains("google.pubsub.v1.Topic.name=" + topicName.toString());
294+
assertThat(bout.toString()).contains(streamArn);
295+
assertThat(bout.toString()).contains(consumerArn);
296+
assertThat(bout.toString()).contains(awsRoleArn);
297+
assertThat(bout.toString()).contains(gcpServiceAccount);
298+
276299
bout.reset();
277300
// Test delete topic.
278301
DeleteTopicExample.deleteTopicExample(projectId, topicId);
279302
assertThat(bout.toString()).contains("Deleted topic.");
303+
304+
bout.reset();
305+
// Test create topic with Kinesis ingestion settings.
306+
CreateTopicWithKinesisIngestionExample.createTopicWithKinesisIngestionExample(
307+
projectId, ingestionTopicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
308+
assertThat(bout.toString())
309+
.contains("google.pubsub.v1.Topic.name=" + ingestionTopicName.toString());
310+
assertThat(bout.toString()).contains(streamArn);
311+
assertThat(bout.toString()).contains(consumerArn);
312+
assertThat(bout.toString()).contains(awsRoleArn);
313+
assertThat(bout.toString()).contains(gcpServiceAccount);
314+
315+
bout.reset();
316+
// Test update existing Kinesis ingestion settings.
317+
UpdateTopicTypeExample.updateTopicTypeExample(
318+
projectId, ingestionTopicId, streamArn, consumerArn2, awsRoleArn, gcpServiceAccount);
319+
assertThat(bout.toString())
320+
.contains("google.pubsub.v1.Topic.name=" + ingestionTopicName.toString());
321+
assertThat(bout.toString()).contains(streamArn);
322+
assertThat(bout.toString()).contains(consumerArn2);
323+
assertThat(bout.toString()).contains(awsRoleArn);
324+
assertThat(bout.toString()).contains(gcpServiceAccount);
325+
326+
bout.reset();
327+
// Test delete Kinesis ingestion topic.
328+
DeleteTopicExample.deleteTopicExample(projectId, ingestionTopicId);
329+
assertThat(bout.toString()).contains("Deleted topic.");
280330
}
281331
}

0 commit comments

Comments
 (0)