Skip to content

/AWS1/IF_KNS=>CREATECHANNEL()

About CreateChannel

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.

Method Signature

METHODS /AWS1/IF_KNS~CREATECHANNEL
  IMPORTING
    !IV_CHANNELNAME TYPE /AWS1/KNSCHANNELNAME OPTIONAL
    !IV_SERVICEEXECUTIONROLEARN TYPE /AWS1/KNSROLEARN OPTIONAL
    !IT_STREAMCONFIGURATIONLIST TYPE /AWS1/CL_KNSCHANNELSTREAMCONF=>TT_CHANNELSTREAMCONFLIST OPTIONAL
    !IO_S3DESTINATIONCONF TYPE REF TO /AWS1/CL_KNSS3DESTINATIONCONF OPTIONAL
    !IO_S3TABLESDESTINATIONCONF TYPE REF TO /AWS1/CL_KNSS3TABLESDSTCONF OPTIONAL
    !IO_ENCRYPTIONCONFIGURATION TYPE REF TO /AWS1/CL_KNSCHANNELENCCONF OPTIONAL
    !IT_TAGS TYPE /AWS1/CL_KNSTAGMAP_W=>TT_TAGMAP OPTIONAL
    !IO_LOGGINGCONFIGURATION TYPE REF TO /AWS1/CL_KNSCHANNELLOGGINGCONF OPTIONAL
  RETURNING
    VALUE(OO_OUTPUT) TYPE REF TO /aws1/cl_knscreatechnloutput
  RAISING
    /AWS1/CX_KNSACCESSDENIEDEX
    /AWS1/CX_KNSINVALIDARGUMENTEX
    /AWS1/CX_KNSKMSACCESSDENIEDEX
    /AWS1/CX_KNSKMSDISABLEDEX
    /AWS1/CX_KNSKMSINVALIDSTATEEX
    /AWS1/CX_KNSKMSNOTFOUNDEX
    /AWS1/CX_KNSKMSOPTINREQUIRED
    /AWS1/CX_KNSKMSTHROTTLINGEX
    /AWS1/CX_KNSLIMITEXCEEDEDEX
    /AWS1/CX_KNSRESOURCEINUSEEX
    /AWS1/CX_KNSRESOURCENOTFOUNDEX
    /AWS1/CX_KNSVALIDATIONEX
    /AWS1/CX_KNSCLIENTEXC
    /AWS1/CX_KNSSERVEREXC
    /AWS1/CX_RT_TECHNICAL_GENERIC
    /AWS1/CX_RT_SERVICE_GENERIC.

IMPORTING

Required arguments:

iv_channelname TYPE /AWS1/KNSCHANNELNAME /AWS1/KNSCHANNELNAME

The name of the channel. The name is unique within your Amazon Web Services account and Amazon Web Services Region.

iv_serviceexecutionrolearn TYPE /AWS1/KNSROLEARN /AWS1/KNSROLEARN

The Amazon Resource Name (ARN) of the IAM role that Amazon Kinesis Data Streams assumes to write records to the destination.

it_streamconfigurationlist TYPE /AWS1/CL_KNSCHANNELSTREAMCONF=>TT_CHANNELSTREAMCONFLIST TT_CHANNELSTREAMCONFLIST

The source stream configuration for the channel. Currently, one stream is supported per channel.

Optional arguments:

io_s3destinationconf TYPE REF TO /AWS1/CL_KNSS3DESTINATIONCONF /AWS1/CL_KNSS3DESTINATIONCONF

The configuration for delivery to a general purpose Amazon S3 bucket. Specify this parameter when S3TablesDestinationConfiguration is not specified.

io_s3tablesdestinationconf TYPE REF TO /AWS1/CL_KNSS3TABLESDSTCONF /AWS1/CL_KNSS3TABLESDSTCONF

The configuration for delivery to streaming tables on Apache Iceberg in Amazon S3 Tables. Specify this parameter when S3DestinationConfiguration is not specified.

io_encryptionconfiguration TYPE REF TO /AWS1/CL_KNSCHANNELENCCONF /AWS1/CL_KNSCHANNELENCCONF

The server-side encryption configuration that uses an Amazon Web Services KMS key to encrypt data delivered to the destination.

it_tags TYPE /AWS1/CL_KNSTAGMAP_W=>TT_TAGMAP TT_TAGMAP

A set of key-value pairs to assign to the channel. A tag consists of a required key and an optional value.

io_loggingconfiguration TYPE REF TO /AWS1/CL_KNSCHANNELLOGGINGCONF /AWS1/CL_KNSCHANNELLOGGINGCONF

The Amazon CloudWatch Logs configuration for the channel.

RETURNING

oo_output TYPE REF TO /aws1/cl_knscreatechnloutput /AWS1/CL_KNSCREATECHNLOUTPUT

Examples

Syntax Example

This is an example of the syntax for calling the method. It includes every possible argument and initializes every possible value. The data provided is not necessarily semantically accurate (for example the value "string" may be provided for something that is intended to be an instance ID, or in some cases two arguments may be mutually exclusive). The syntax shows the ABAP syntax for creating the various data structures.

DATA(lo_result) = lo_client->createchannel(
  io_encryptionconfiguration = new /aws1/cl_knschannelencconf(
    iv_encryptiontype = |string|
    iv_keyid = |string|
  )
  io_loggingconfiguration = new /aws1/cl_knschannelloggingconf(
    io_cloudwatchlogs = new /aws1/cl_knscloudwatchlogs(
      iv_enabled = ABAP_TRUE
      iv_loggroupname = |string|
      iv_logstreamname = |string|
    )
  )
  io_s3destinationconf = new /aws1/cl_knss3destinationconf(
    io_deadletterqueues3conf = new /aws1/cl_knsdeadltrqueues3conf(
      iv_bucketarn = |string|
      iv_erroroutputprefix = |string|
      iv_expectedbucketowner = |string|
    )
    io_storageconfiguration = new /aws1/cl_knss3storageconf(
      iv_bucketarn = |string|
      iv_compressiontype = |string|
      iv_expectedbucketowner = |string|
      iv_outputkeytemplate = |string|
      iv_storageclass = |string|
    )
    iv_datafreshnessinseconds = 123
  )
  io_s3tablesdestinationconf = new /aws1/cl_knss3tablesdstconf(
    io_deadletterqueues3conf = new /aws1/cl_knsdeadltrqueues3conf(
      iv_bucketarn = |string|
      iv_erroroutputprefix = |string|
      iv_expectedbucketowner = |string|
    )
    it_s3tablesconfigurationlist = VALUE /aws1/cl_knss3tablesconf=>tt_s3tablesconfigurationlist(
      (
        new /aws1/cl_knss3tablesconf(
          io_partitionspec = new /aws1/cl_knspartitionspec(
            it_partitionfields = VALUE /aws1/cl_knspartitionfield=>tt_partitionfieldlist(
              (
                new /aws1/cl_knspartitionfield(
                  iv_sourcename = |string|
                  iv_transform = |string|
                )
              )
            )
          )
          iv_compressiontype = |string|
          iv_namespace = |string|
          iv_tablebucketarn = |string|
          iv_tablename = |string|
        )
      )
    )
    iv_datafreshnessinseconds = 123
  )
  it_streamconfigurationlist = VALUE /aws1/cl_knschannelstreamconf=>tt_channelstreamconflist(
    (
      new /aws1/cl_knschannelstreamconf(
        io_recordconfiguration = new /aws1/cl_knsrecordconf(
          iv_gsrschemaarn = |string|
          iv_recordformattype = |string|
        )
        iv_streamarn = |string|
      )
    )
  )
  it_tags = VALUE /aws1/cl_knstagmap_w=>tt_tagmap(
    (
      VALUE /aws1/cl_knstagmap_w=>ts_tagmap_maprow(
        key = |string|
        value = new /aws1/cl_knstagmap_w( |string| )
      )
    )
  )
  iv_channelname = |string|
  iv_serviceexecutionrolearn = |string|
).

This is an example of reading all possible response values

lo_result = lo_result.
IF lo_result IS NOT INITIAL.
  lo_channeldescription = lo_result->get_channeldescription( ).
  IF lo_channeldescription IS NOT INITIAL.
    lv_channelname = lo_channeldescription->get_channelname( ).
    lv_channelarn = lo_channeldescription->get_channelarn( ).
    lv_channelid = lo_channeldescription->get_channelid( ).
    lv_channelstatus = lo_channeldescription->get_channelstatus( ).
    lv_channelstatusreason = lo_channeldescription->get_channelstatusreason( ).
    lv_timestamp = lo_channeldescription->get_channelcreationtimestamp( ).
    lv_rolearn = lo_channeldescription->get_serviceexecutionrolearn( ).
    LOOP AT lo_channeldescription->get_streamconfigurationlist( ) into lo_row.
      lo_row_1 = lo_row.
      IF lo_row_1 IS NOT INITIAL.
        lv_streamarn = lo_row_1->get_streamarn( ).
        lv_timestamp = lo_row_1->get_streamcreationtimestamp( ).
        lo_recordconfiguration = lo_row_1->get_recordconfiguration( ).
        IF lo_recordconfiguration IS NOT INITIAL.
          lv_recordformattype = lo_recordconfiguration->get_recordformattype( ).
          lv_gsrschemaarn = lo_recordconfiguration->get_gsrschemaarn( ).
        ENDIF.
      ENDIF.
    ENDLOOP.
    lo_s3destinationdescriptio = lo_channeldescription->get_s3destinationconf( ).
    IF lo_s3destinationdescriptio IS NOT INITIAL.
      lv_datafreshnessinseconds = lo_s3destinationdescriptio->get_datafreshnessinseconds( ).
      lo_deadletterqueues3config = lo_s3destinationdescriptio->get_deadletterqueues3conf( ).
      IF lo_deadletterqueues3config IS NOT INITIAL.
        lv_bucketarn = lo_deadletterqueues3config->get_bucketarn( ).
        lv_expectedbucketowner = lo_deadletterqueues3config->get_expectedbucketowner( ).
        lv_s3erroroutputprefix = lo_deadletterqueues3config->get_erroroutputprefix( ).
      ENDIF.
      lo_s3storageconfiguration = lo_s3destinationdescriptio->get_storageconfiguration( ).
      IF lo_s3storageconfiguration IS NOT INITIAL.
        lv_bucketarn = lo_s3storageconfiguration->get_bucketarn( ).
        lv_expectedbucketowner = lo_s3storageconfiguration->get_expectedbucketowner( ).
        lv_s3outputkeytemplate = lo_s3storageconfiguration->get_outputkeytemplate( ).
        lv_s3storageclass = lo_s3storageconfiguration->get_storageclass( ).
        lv_s3compressiontype = lo_s3storageconfiguration->get_compressiontype( ).
      ENDIF.
    ENDIF.
    lo_s3tablesdestinationdesc = lo_channeldescription->get_s3tablesdestinationconf( ).
    IF lo_s3tablesdestinationdesc IS NOT INITIAL.
      lv_datafreshnessinseconds = lo_s3tablesdestinationdesc->get_datafreshnessinseconds( ).
      lo_deadletterqueues3config = lo_s3tablesdestinationdesc->get_deadletterqueues3conf( ).
      IF lo_deadletterqueues3config IS NOT INITIAL.
        lv_bucketarn = lo_deadletterqueues3config->get_bucketarn( ).
        lv_expectedbucketowner = lo_deadletterqueues3config->get_expectedbucketowner( ).
        lv_s3erroroutputprefix = lo_deadletterqueues3config->get_erroroutputprefix( ).
      ENDIF.
      LOOP AT lo_s3tablesdestinationdesc->get_s3tablesconflist( ) into lo_row_2.
        lo_row_3 = lo_row_2.
        IF lo_row_3 IS NOT INITIAL.
          lv_tablebucketarn = lo_row_3->get_tablebucketarn( ).
          lv_s3tablesnamespace = lo_row_3->get_namespace( ).
          lv_s3tablestablename = lo_row_3->get_tablename( ).
          lv_s3tablescompressiontype = lo_row_3->get_compressiontype( ).
          lo_partitionspec = lo_row_3->get_partitionspec( ).
          IF lo_partitionspec IS NOT INITIAL.
            LOOP AT lo_partitionspec->get_partitionfields( ) into lo_row_4.
              lo_row_5 = lo_row_4.
              IF lo_row_5 IS NOT INITIAL.
                lv_partitiontransform = lo_row_5->get_transform( ).
                lv_partitionsourcename = lo_row_5->get_sourcename( ).
              ENDIF.
            ENDLOOP.
          ENDIF.
        ENDIF.
      ENDLOOP.
    ENDIF.
    lo_channelencryptionconfig = lo_channeldescription->get_encryptionconfiguration( ).
    IF lo_channelencryptionconfig IS NOT INITIAL.
      lv_channelencryptiontype = lo_channelencryptionconfig->get_encryptiontype( ).
      lv_keyid = lo_channelencryptionconfig->get_keyid( ).
    ENDIF.
    lo_channelloggingconfigura = lo_channeldescription->get_loggingconfiguration( ).
    IF lo_channelloggingconfigura IS NOT INITIAL.
      lo_cloudwatchlogs = lo_channelloggingconfigura->get_cloudwatchlogs( ).
      IF lo_cloudwatchlogs IS NOT INITIAL.
        lv_booleanobject = lo_cloudwatchlogs->get_enabled( ).
        lv_cloudwatchloggroupname = lo_cloudwatchlogs->get_loggroupname( ).
        lv_cloudwatchlogstreamname = lo_cloudwatchlogs->get_logstreamname( ).
      ENDIF.
    ENDIF.
  ENDIF.
ENDIF.

To create an S3 channel

To create an S3 channel

DATA(lo_result) = lo_client->createchannel(
  io_encryptionconfiguration = new /aws1/cl_knschannelencconf(
    iv_encryptiontype = |KMS|
    iv_keyid = |arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab|
  )
  io_loggingconfiguration = new /aws1/cl_knschannelloggingconf(
    io_cloudwatchlogs = new /aws1/cl_knscloudwatchlogs(
      iv_enabled = ABAP_TRUE
      iv_loggroupname = |/aws/kinesis/my-channel|
      iv_logstreamname = |my-channel-log-stream|
    )
  )
  io_s3destinationconf = new /aws1/cl_knss3destinationconf(
    io_deadletterqueues3conf = new /aws1/cl_knsdeadltrqueues3conf(
      iv_bucketarn = |arn:aws:s3:::my-channel-dlq-bucket|
      iv_expectedbucketowner = |123456789012|
    )
    io_storageconfiguration = new /aws1/cl_knss3storageconf(
      iv_bucketarn = |arn:aws:s3:::my-channel-bucket|
      iv_compressiontype = |ZSTD|
      iv_expectedbucketowner = |123456789012|
    )
  )
  it_streamconfigurationlist = VALUE /aws1/cl_knschannelstreamconf=>tt_channelstreamconflist(
    (
      new /aws1/cl_knschannelstreamconf(
        io_recordconfiguration = new /aws1/cl_knsrecordconf( iv_recordformattype = |JSON| )
        iv_streamarn = |arn:aws:kinesis:us-east-1:123456789012:stream/my-stream-name|
      )
    )
  )
  iv_channelname = |my-channel-name|
  iv_serviceexecutionrolearn = |arn:aws:iam::123456789012:role/my-channel-role|
).

To create an S3 Tables channel

To create an S3 Tables channel

DATA(lo_result) = lo_client->createchannel(
  io_encryptionconfiguration = new /aws1/cl_knschannelencconf(
    iv_encryptiontype = |KMS|
    iv_keyid = |arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab|
  )
  io_loggingconfiguration = new /aws1/cl_knschannelloggingconf(
    io_cloudwatchlogs = new /aws1/cl_knscloudwatchlogs(
      iv_enabled = ABAP_TRUE
      iv_loggroupname = |/aws/kinesis/my-channel|
      iv_logstreamname = |my-channel-log-stream|
    )
  )
  io_s3tablesdestinationconf = new /aws1/cl_knss3tablesdstconf(
    io_deadletterqueues3conf = new /aws1/cl_knsdeadltrqueues3conf(
      iv_bucketarn = |arn:aws:s3:::my-channel-dlq-bucket|
      iv_expectedbucketowner = |123456789012|
    )
    it_s3tablesconfigurationlist = VALUE /aws1/cl_knss3tablesconf=>tt_s3tablesconfigurationlist(
      (
        new /aws1/cl_knss3tablesconf(
          io_partitionspec = new /aws1/cl_knspartitionspec(
            it_partitionfields = VALUE /aws1/cl_knspartitionfield=>tt_partitionfieldlist(
              (
                new /aws1/cl_knspartitionfield(
                  iv_sourcename = |creation_ts|
                  iv_transform = |TIME_HOUR|
                )
              )
            )
          )
          iv_compressiontype = |ZSTD|
          iv_namespace = |my_namespace|
          iv_tablebucketarn = |arn:aws:s3tables:us-east-1:123456789012:bucket/my-table-bucket|
          iv_tablename = |my_table|
        )
      )
    )
  )
  it_streamconfigurationlist = VALUE /aws1/cl_knschannelstreamconf=>tt_channelstreamconflist(
    (
      new /aws1/cl_knschannelstreamconf(
        io_recordconfiguration = new /aws1/cl_knsrecordconf(
          iv_gsrschemaarn = |arn:aws:glue:us-east-1:123456789012:schema/my-registry/my-schema|
          iv_recordformattype = |JSON|
        )
        iv_streamarn = |arn:aws:kinesis:us-east-1:123456789012:stream/my-stream-name|
      )
    )
  )
  iv_channelname = |my-channel-name|
  iv_serviceexecutionrolearn = |arn:aws:iam::123456789012:role/my-channel-role|
).