docs: Add Kinesis ingestion samples

README.md

@@ -257,6 +257,7 @@ Samples are in the [`samples/`](https://github.com/googleapis/java-pubsub/tree/m
 | 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) |
 | 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) |
 | 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) |
+| 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) |
 | 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) |
 | 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) |
 | 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
 | 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) |
 | 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) |
 | 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) |
+| 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) |
 | 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) |
 | 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) |
 | 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

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package pubsub;
+
+// [START pubsub_create_topic_with_kinesis_ingestion]
+
+import com.google.cloud.pubsub.v1.TopicAdminClient;
+import com.google.pubsub.v1.IngestionDataSourceSettings;
+import com.google.pubsub.v1.Topic;
+import com.google.pubsub.v1.TopicName;
+import java.io.IOException;
+
+public class CreateTopicWithKinesisIngestionExample {
+  public static void main(String... args) throws Exception {
+    // TODO(developer): Replace these variables before running the sample.
+    String projectId = "your-project-id";
+    String topicId = "your-topic-id";
+    // Kinesis ingestion settings.
+    String streamArn = "stream-arn";
+    String consumerArn = "consumer-arn";
+    String awsRoleArn = "aws-role-arn";
+    String gcpServiceAccount = "gcp-service-account";
+
+    createTopicWithKinesisIngestionExample(
+        projectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
+  }
+
+  public static void createTopicWithKinesisIngestionExample(
+      String projectId,
+      String topicId,
+      String streamArn,
+      String consumerArn,
+      String awsRoleArn,
+      String gcpServiceAccount)
+      throws IOException {
+    try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
+      TopicName topicName = TopicName.of(projectId, topicId);
+
+      IngestionDataSourceSettings.AwsKinesis awsKinesis =
+          IngestionDataSourceSettings.AwsKinesis.newBuilder()
+              .setStreamArn(streamArn)
+              .setConsumerArn(consumerArn)
+              .setAwsRoleArn(awsRoleArn)
+              .setGcpServiceAccount(gcpServiceAccount)
+              .build();
+      IngestionDataSourceSettings ingestionDataSourceSettings =
+          IngestionDataSourceSettings.newBuilder().setAwsKinesis(awsKinesis).build();
+
+      Topic topic =
+          topicAdminClient.createTopic(
+              Topic.newBuilder()
+                  .setName(topicName.toString())
+                  .setIngestionDataSourceSettings(ingestionDataSourceSettings)
+                  .build());
+
+      System.out.println("Created topic with Kinesis ingestion settings: " + topic.getAllFields());
+    }
+  }
+}
+// [END pubsub_create_topic_with_kinesis_ingestion]

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

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package pubsub;
+
+// [START pubsub_update_topic_type]
+
+import com.google.cloud.pubsub.v1.TopicAdminClient;
+import com.google.protobuf.FieldMask;
+import com.google.pubsub.v1.IngestionDataSourceSettings;
+import com.google.pubsub.v1.Topic;
+import com.google.pubsub.v1.TopicName;
+import com.google.pubsub.v1.UpdateTopicRequest;
+import java.io.IOException;
+
+public class UpdateTopicTypeExample {
+  public static void main(String... args) throws Exception {
+    // TODO(developer): Replace these variables before running the sample.
+    String projectId = "your-project-id";
+    String topicId = "your-topic-id";
+    // Kinesis ingestion settings.
+    String streamArn = "stream-arn";
+    String consumerArn = "consumer-arn";
+    String awsRoleArn = "aws-role-arn";
+    String gcpServiceAccount = "gcp-service-account";
+
+    UpdateTopicTypeExample.updateTopicTypeExample(
+        projectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
+  }
+
+  public static void updateTopicTypeExample(
+      String projectId,
+      String topicId,
+      String streamArn,
+      String consumerArn,
+      String awsRoleArn,
+      String gcpServiceAccount)
+      throws IOException {
+    try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
+      TopicName topicName = TopicName.of(projectId, topicId);
+
+      IngestionDataSourceSettings.AwsKinesis awsKinesis =
+          IngestionDataSourceSettings.AwsKinesis.newBuilder()
+              .setStreamArn(streamArn)
+              .setConsumerArn(consumerArn)
+              .setAwsRoleArn(awsRoleArn)
+              .setGcpServiceAccount(gcpServiceAccount)
+              .build();
+      IngestionDataSourceSettings ingestionDataSourceSettings =
+          IngestionDataSourceSettings.newBuilder().setAwsKinesis(awsKinesis).build();
+
+      // Construct the topic with Kinesis ingestion settings.
+      Topic topic =
+          Topic.newBuilder()
+              .setName(topicName.toString())
+              .setIngestionDataSourceSettings(ingestionDataSourceSettings)
+              .build();
+
+      // Construct a field mask to indicate which field to update in the topic.
+      FieldMask updateMask =
+          FieldMask.newBuilder().addPaths("ingestion_data_source_settings").build();
+
+      UpdateTopicRequest request =
+          UpdateTopicRequest.newBuilder().setTopic(topic).setUpdateMask(updateMask).build();
+
+      Topic response = topicAdminClient.updateTopic(request);
+
+      System.out.println(
+          "Updated topic with Kinesis ingestion settings: " + response.getAllFields());
+    }
+  }
+}
+// [END pubsub_update_topic_type]

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

@@ -52,6 +52,7 @@ public class AdminIT {
   private static final String projectId = System.getenv("GOOGLE_CLOUD_PROJECT");
   private static final String _suffix = UUID.randomUUID().toString();
   private static final String topicId = "iam-topic-" + _suffix;
+  private static final String ingestionTopicId = "ingestion-topic-" + _suffix;
   private static final String pullSubscriptionId = "iam-pull-subscription-" + _suffix;
   private static final String pushSubscriptionId = "iam-push-subscription-" + _suffix;
   private static final String orderedSubscriptionId = "iam-ordered-subscription-" + _suffix;
@@ -63,8 +64,20 @@ public class AdminIT {
       "java_samples_data_set" + _suffix.replace("-", "_");
   private static final String bigquerySubscriptionId = "iam-bigquery-subscription-" + _suffix;
   private static final String bigqueryTableId = "java_samples_table_" + _suffix;
+  private static final String streamArn =
+      "arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name";
+  private static final String consumerArn =
+      "arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name/"
+          + "consumer/consumer-1:1111111111";
+  private static final String consumerArn2 =
+      "arn:aws:kinesis:us-west-2:111111111111:stream/fake-stream-name/"
+          + "consumer/consumer-2:2222222222";
+  private static final String awsRoleArn = "arn:aws:iam::111111111111:role/fake-role-name";
+  private static final String gcpServiceAccount =
+      "[email protected]";
 
   private static final TopicName topicName = TopicName.of(projectId, topicId);
+  private static final TopicName ingestionTopicName = TopicName.of(projectId, ingestionTopicId);
   private static final SubscriptionName pullSubscriptionName =
       SubscriptionName.of(projectId, pullSubscriptionId);
   private static final SubscriptionName pushSubscriptionName =
@@ -273,9 +286,46 @@ public void testAdmin() throws Exception {
     DeleteSubscriptionExample.deleteSubscriptionExample(projectId, bigquerySubscriptionId);
     assertThat(bout.toString()).contains("Deleted subscription.");
 
+    bout.reset();
+    // Update topic type to Kinesis ingestion.
+    UpdateTopicTypeExample.updateTopicTypeExample(
+        projectId, topicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
+    assertThat(bout.toString()).contains("google.pubsub.v1.Topic.name=" + topicName.toString());
+    assertThat(bout.toString()).contains(streamArn);
+    assertThat(bout.toString()).contains(consumerArn);
+    assertThat(bout.toString()).contains(awsRoleArn);
+    assertThat(bout.toString()).contains(gcpServiceAccount);
+
     bout.reset();
     // Test delete topic.
     DeleteTopicExample.deleteTopicExample(projectId, topicId);
     assertThat(bout.toString()).contains("Deleted topic.");
+
+    bout.reset();
+    // Test create topic with Kinesis ingestion settings.
+    CreateTopicWithKinesisIngestionExample.createTopicWithKinesisIngestionExample(
+        projectId, ingestionTopicId, streamArn, consumerArn, awsRoleArn, gcpServiceAccount);
+    assertThat(bout.toString())
+        .contains("google.pubsub.v1.Topic.name=" + ingestionTopicName.toString());
+    assertThat(bout.toString()).contains(streamArn);
+    assertThat(bout.toString()).contains(consumerArn);
+    assertThat(bout.toString()).contains(awsRoleArn);
+    assertThat(bout.toString()).contains(gcpServiceAccount);
+
+    bout.reset();
+    // Test update existing Kinesis ingestion settings.
+    UpdateTopicTypeExample.updateTopicTypeExample(
+        projectId, ingestionTopicId, streamArn, consumerArn2, awsRoleArn, gcpServiceAccount);
+    assertThat(bout.toString())
+        .contains("google.pubsub.v1.Topic.name=" + ingestionTopicName.toString());
+    assertThat(bout.toString()).contains(streamArn);
+    assertThat(bout.toString()).contains(consumerArn2);
+    assertThat(bout.toString()).contains(awsRoleArn);
+    assertThat(bout.toString()).contains(gcpServiceAccount);
+
+    bout.reset();
+    // Test delete Kinesis ingestion topic.
+    DeleteTopicExample.deleteTopicExample(projectId, ingestionTopicId);
+    assertThat(bout.toString()).contains("Deleted topic.");
   }
 }