kinesis/aws.sdk.kotlin.services.kinesis/KinesisClient/createChannel

createChannel

abstract suspend fun createChannel(input: CreateChannelRequest): CreateChannelResponse

Creates a channel that delivers records from a Kinesis data stream to a destination. A channel reads records from the specified stream and writes them to streaming tables on Apache Iceberg (Amazon S3 Tables) or to a general purpose Amazon S3 bucket.

You must specify either S3DestinationConfiguration or S3TablesDestinationConfiguration, but not both.

To use this operation, you must have permission to pass the specified service execution IAM role to Amazon Kinesis Data Streams (the iam:PassRole permission on that role).

Creating a channel is an asynchronous operation. Upon receiving the request, Amazon Kinesis Data Streams returns immediately with the channel in the CREATING state. After provisioning is complete, Amazon Kinesis Data Streams sets the state to ACTIVE. You can use DescribeChannel to check the current state.

This operation is only supported for data streams with the on-demand capacity mode.

This operation has a call limit of 5 transactions per second (TPS) for each Amazon Web Services account. Exceeding 5 TPS results in a LimitExceededException.

Samples

// To create an S3 channel
val resp = kinesisClient.createChannel {
    channelName = "my-channel-name"
    serviceExecutionRoleArn = "arn:aws:iam::123456789012:role/my-channel-role"
    streamConfigurationList = listOf<ChannelStreamConfiguration>(
        ChannelStreamConfiguration {
            streamArn = "arn:aws:kinesis:us-east-1:123456789012:stream/my-stream-name"
            recordConfiguration = RecordConfiguration {
                recordFormatType = RecordFormatType.fromValue("JSON")
            }
        }            
    )
    s3DestinationConfiguration = S3DestinationConfiguration {
        storageConfiguration = S3StorageConfiguration {
            bucketArn = "arn:aws:s3:::my-channel-bucket"
            expectedBucketOwner = "123456789012"
            compressionType = S3CompressionType.fromValue("ZSTD")
        }
        deadLetterQueueS3Configuration = DeadLetterQueueS3Configuration {
            bucketArn = "arn:aws:s3:::my-channel-dlq-bucket"
            expectedBucketOwner = "123456789012"
        }
    }
    encryptionConfiguration = ChannelEncryptionConfiguration {
        encryptionType = ChannelEncryptionType.fromValue("KMS")
        keyId = "arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab"
    }
    loggingConfiguration = ChannelLoggingConfiguration {
        cloudWatchLogs = CloudWatchLogs {
            enabled = true
            logGroupName = "/aws/kinesis/my-channel"
            logStreamName = "my-channel-log-stream"
        }
    }
}

// To create an S3 Tables channel
val resp = kinesisClient.createChannel {
    channelName = "my-channel-name"
    serviceExecutionRoleArn = "arn:aws:iam::123456789012:role/my-channel-role"
    streamConfigurationList = listOf<ChannelStreamConfiguration>(
        ChannelStreamConfiguration {
            streamArn = "arn:aws:kinesis:us-east-1:123456789012:stream/my-stream-name"
            recordConfiguration = RecordConfiguration {
                recordFormatType = RecordFormatType.fromValue("JSON")
                gsrSchemaArn = "arn:aws:glue:us-east-1:123456789012:schema/my-registry/my-schema"
            }
        }            
    )
    s3TablesDestinationConfiguration = S3TablesDestinationConfiguration {
        deadLetterQueueS3Configuration = DeadLetterQueueS3Configuration {
            bucketArn = "arn:aws:s3:::my-channel-dlq-bucket"
            expectedBucketOwner = "123456789012"
        }
        s3TablesConfigurationList = listOf<S3TablesConfiguration>(
            S3TablesConfiguration {
                tableBucketArn = "arn:aws:s3tables:us-east-1:123456789012:bucket/my-table-bucket"
                namespace = "my_namespace"
                tableName = "my_table"
                compressionType = S3TablesCompressionType.fromValue("ZSTD")
                partitionSpec = PartitionSpec {
                    partitionFields = listOf<PartitionField>(
                        PartitionField {
                            transform = PartitionTransform.fromValue("TIME_HOUR")
                            sourceName = "creation_ts"
                        }                            
                    )
                }
            }                
        )
    }
    encryptionConfiguration = ChannelEncryptionConfiguration {
        encryptionType = ChannelEncryptionType.fromValue("KMS")
        keyId = "arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab"
    }
    loggingConfiguration = ChannelLoggingConfiguration {
        cloudWatchLogs = CloudWatchLogs {
            enabled = true
            logGroupName = "/aws/kinesis/my-channel"
            logStreamName = "my-channel-log-stream"
        }
    }
}