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());

}

}