- Configuration and Testing
- Component Configurations
- Post Processing XSL Configuration
- Validate Input
- Error handling configuration
- AWS Connection Details
- Kinesis Stream Name
- Application Name
- Initial Position In Stream
- Payload Data Type
- Check Point Interval (ms)
- Number of Retries
- Back Off Time (ms)
- Threadpool Configuration
- Component Configurations
- Functional Demonstration
KinesisProducer microservice is used to consume data records to Amazon Kinesis Streams.
Configuration and Testing
The attributes can be configured in the Configuration Properties Sheet (CPS) as below.
Figure 1: Component Configuration Properties
Post Processing XSL Configuration
Post Processing XSL configuration can be used to transform response message before sending it to the output port.
If this attribute is enabled, the service tries to validate the input received. If disabled, service will not validate the input. For more details, refer Validate Input section under Interaction Configurations in Common Configurations page.
Error handling configuration
The remedial actions to be taken when a particular error occurs can be configured using this attribute.
Click the ellipsis button against this property to configure Error Handling properties for different types of Errors. By default, the options Log to error logs, Stop service and Send to error port are enabled.
AWS Connection Details
Click the AWS Connection Details ellipsis button to configure the properties.
Specify the Access Key id
Specify the Secret Key
The region of AWS Management console.
Kinesis Stream Name
Name of the Kinesis Stream from which the data records are read.
Kinesis Consumer Library uses a unique Amazon Dynamo DB Table to keep track of the application state. It uses the application name to create the table; each Application Name must be unique.
Initial Position In Stream
Used to specify the position in the stream where a new application should start from.
- LATEST: Fetches you always read the most recent data in the shard
- TRIM_HORIZON: Gets the record of last untrimmed record in the shard in the system (the oldest data record in the shard)
Scenario: Send 10 data records to the stream when Consumer is inactive.
Now starting the consumer in LATEST mode will not fetch you any records.
If the consumer is in TRIM_HORIZON mode, all the unread messages(10 in this case) are retrieved from the stream.
Payload Data Type
Data blob is the data of interest your data producer adds to a stream. The maximum size of a data blob is 1 megabyte (MB) .The base64 encoded Data blob is called Payload Data Type
Specify the required output type. Data is interpreted as UTF-8 characters if string is selected.
Check Point Interval (ms)
Check Point Interval in milliseconds is the interval at the state of the streams are internally stored in Dynamo Table
Number of Retries
Number of times the Service should try for checkpointing / Processing records in whichever case an exception occurs.
Back Off Time (ms)
The interval between the consecutive retries of checkpointing / Processing records in whichever case an exception occurs.
This property is used when there is a need to process messages in parallel within the component, still maintaining the sequence from the external perspective.
Click the Threadpool Configuration ellipsis button to configure the Threadpool Configuration properties.
Enable Thread Pool
Enable this option to configure the properties that appear as below.
Number of requests to be processed in parallel within the component. Default value is '1'.
Batch Eviction Interval (in ms)
Time in milliseconds after which the threads are evicted in case of inactivity. New threads are created in place of evicted threads when new requests are received. Default value is '1000'.
The following flow demonstrates a flow of Kinesis Consumer ,which fetches the data from the provided StreamName
The output message consists of following elements
Data retrieved from the stream
A partition key is used to group data by shard within a stream. The Streams service segregates the data records belonging to a stream into multiple shards, using the partition key associated with each data record to determine which shard a given data record belongs to.
Sequence number defines the sequence of Records in a particular shard
Each data record has a unique sequence number.
Approximate time stamp of the record after retrieving from the kinesis stream.