9.2.2 Amazon Kinesis

The Kinesis Streams Handler streams data to applications hosted on the Amazon Cloud or in your environment.

This chapter describes how to use the Kinesis Streams Handler.

• Overview
  
• Detailed Functionality
  
• Setting Up and Running the Kinesis Streams Handler
  
• Kinesis Handler Performance Considerations
  
• Troubleshooting
  

9.2.2.1 Overview

Amazon Kinesis is a messaging system that is hosted in the Amazon Cloud. Kinesis streams can be used to stream data to other Amazon Cloud applications such as Amazon S3 and Amazon Redshift. Using the Kinesis Streams Handler, you can also stream data to applications hosted on the Amazon Cloud or at your site. Amazon Kinesis streams provides functionality similar to Apache Kafka.

The logical concepts map is as follows:

• Kafka Topics = Kinesis Streams

• Kafka Partitions = Kinesis Shards

A Kinesis stream must have at least one shard.

9.2.2.2 Detailed Functionality

• Amazon Kinesis Java SDK
  
• Kinesis Streams Input Limits
  

9.2.2.2.1 Amazon Kinesis Java SDK

The Oracle GoldenGate Kinesis Streams Handler uses the AWS Kinesis Java SDK to push data to Amazon Kinesis, see Amazon Kinesis Streams Developer Guide at:

http://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html.

The Kinesis Steams Handler was designed and tested with the latest AWS Kinesis Java SDK version 1.11.107. These are the dependencies:

• Group ID: com.amazonaws

• Artifact ID: aws-java-sdk-kinesis

• Version: 1.11.107

Oracle GoldenGate for Distributed Applications and Analytics (GG for DAA) does not ship with the AWS Kinesis Java SDK. Oracle recommends that you use the AWS Kinesis Java SDK identified in the Certification Matrix, see GoldenGate Certifications.

Note:

It is assumed by moving to the latest AWS Kinesis Java SDK that there are no changes to the interface, which can break compatibility with the Kinesis Streams Handler.

You can download the AWS Java SDK, including Kinesis from:

https://aws.amazon.com/sdk-for-java/

9.2.2.2.2 Kinesis Streams Input Limits

The upper input limit for a Kinesis stream with a single shard is 1000 messages per second up to a total data size of 1MB per second. Adding streams or shards can increase the potential throughput such as the following:

• 1 stream with 2 shards = 2000 messages per second up to a total data size of 2MB per second

• 3 streams of 1 shard each = 3000 messages per second up to a total data size of 3MB per second

The scaling that you can achieve with the Kinesis Streams Handler depends on how you configure the handler. Kinesis stream names are resolved at runtime based on the configuration of the Kinesis Streams Handler.

Shards are selected by the hash the partition key. The partition key for a Kinesis message cannot be null or an empty string (""). A null or empty string partition key results in a Kinesis error that results in an abend of the Replicat process.

Maximizing throughput requires that the Kinesis Streams Handler configuration evenly distributes messages across streams and shards.

To achieve the best distribution across shards in a Kinesis stream, select a partitioning key which rapidly changes. You can select ${primaryKeys} as it is unique per row in the source database. Additionally, operations for the same row are sent to the same Kinesis stream and shard. When the DEBUG logging is enabled, the Kinesis stream name, sequence number, and the shard number are logged to the log file for successfully sent messages.

9.2.2.3 Setting Up and Running the Kinesis Streams Handler

Instructions for configuring the Kinesis Streams Handler components and running the handler are described in the following sections.

Use the following steps to set up the Kinesis Streams Handler:

1. Create an Amazon AWS account at https://aws.amazon.com/.
2. Log into Amazon AWS.
3. From the main page, select Kinesis (under the Analytics subsection).
4. Select Amazon Kinesis Streams Go to Streams to create Amazon Kinesis streams and shards within streams.
5. Create a client ID and secret to access Kinesis.

   The Kinesis Streams Handler requires these credentials at runtime to successfully connect to Kinesis.

6. Create the client ID and secret:
   1. Select your name in AWS (upper right), and then in the list select My Security Credentials.
   2. Select Access Keys to create and manage access keys.

      Note your client ID and secret upon creation.

      The client ID and secret can only be accessed upon creation. If lost, you have to delete the access key, and then recreate it.

• Set the Classpath in Kinesis Streams Handler
  
• Kinesis Streams Handler Configuration
  
• Using Templates to Resolve the Stream Name and Partition Name
  
• Resolving AWS Credentials
  
• Configuring the Proxy Server for Kinesis Streams Handler
  
• Configuring Security in Kinesis Streams Handler
  

9.2.2.3.1 Set the Classpath in Kinesis Streams Handler

You must configure the gg.classpath property in the Java Adapter properties file to specify the JARs for the AWS Kinesis Java SDK as follows:

gg.classpath={download_dir}/aws-java-sdk-1.11.107/lib/*:{download_dir}/aws-java-sdk-1.11.107/third-party/lib/*

9.2.2.3.2 Kinesis Streams Handler Configuration

You configure the Kinesis Streams Handler operation using the properties file. These properties are located in the Java Adapter properties file (not in the Replicat properties file).

To enable the selection of the Kinesis Streams Handler, you must first configure the handler type by specifying gg.handler.name.type=kinesis_streams and the other Kinesis Streams properties as follows:

Table 9-2 Kinesis Streams Handler Configuration Properties

Properties	Required/ Optional	Legal Values	Default	Explanation
gg.handler.name.type	Required	kinesis_streams	None	Selects the Kinesis Streams Handler for streaming change data capture into Kinesis.
gg.handler.name.mode	Optional	op or tx	op	Choose the operating mode.
gg.handler.name.region	Required	The Amazon region name which is hosting your Kinesis instance.	None	Setting of the Amazon AWS region name is required.
gg.handler.name.proxyServer	Optional	The host name of the proxy server.	None	Set the host name of the proxy server if connectivity to AWS is required to go through a proxy server.
gg.handler.name.proxyPort	Optional	The port number of the proxy server.	None	Set the port name of the proxy server if connectivity to AWS is required to go through a proxy server.
gg.handler.name.proxyUsername	Optional	The username of the proxy server (if credentials are required).	None	Set the username of the proxy server if connectivity to AWS is required to go through a proxy server and the proxy server requires credentials.
gg.handler.name.proxyPassword	Optional	The password of the proxy server (if credentials are required).	None	Set the password of the proxy server if connectivity to AWS is required to go through a proxy server and the proxy server requires credentials.
gg.handler.name.deferFlushAtTxCommit	Optional	true | false	false	When set to false, the Kinesis Streams Handler will flush data to Kinesis at transaction commit for write durability. However, it may be preferable to defer the flush beyond the transaction commit for performance purposes, see Kinesis Handler Performance Considerations.
gg.handler.name.deferFlushOpCount	Optional	Integer	None	Only applicable if gg.handler.name.deferFlushAtTxCommit is set to true. This parameter marks the minimum number of operations that must be received before triggering a flush to Kinesis. Once this number of operations are received, a flush will occur on the next transaction commit and all outstanding operations will be moved from the Kinesis Streams Handler to AWS Kinesis.
gg.handler.name.formatPerOp	Optional	true | false	true	When set to true, it will send messages to Kinesis, once per operation (insert, delete, update). When set to false, operations messages will be concatenated for all the operations and a single message will be sent at the transaction level. Kinesis has a limitation of 1MB max massage size. If 1MB is exceeded then transaction level message will be broken up into multiple messages.
gg.handler.name.customMessageGrouper	Optional	oracle.goldengate.handler.kinesis.KinesisJsonTxMessageGrouper	None	This configuration parameter provides the ability to group Kinesis messages using custom logic. Only one implementation is included in the distribution at this time. The oracle.goldengate.handler.kinesis.KinesisJsonTxMessageGrouperis a custom message which groups JSON operation messages representing operations into a wrapper JSON message that encompasses the transaction. Setting of this value overrides the setting of the gg.handler.formatPerOp setting. Using this feature assumes that the customer is using the JSON formatter (that is gg.handler.name.format=json).
gg.handler.name.streamMappingTemplate	Required	A template string value to resolve the Kinesis message partition key (message key) at runtime.	None	See Using Templates to Resolve the Stream Name and Partition Name for more information.
gg.handler.name.partitionMappingTemplate	Required	A template string value to resolve the Kinesis message partition key (message key) at runtime.	None	See Using Templates to Resolve the Stream Name and Partition Name for more information.
gg.hander.name.format	Required	Any supported pluggable formatter.	delimitedtext | json | json_row | xml | avro_row | avro_opt	Selects the operations message formatter. JSON is likely the best fit for Kinesis.
gg.hander.name.enableStreamCreation	Optional	true	true | false	By default, the Kinesis Handler automatically creates Kinesis streams if they do not already exist. Set to false to disable to automatic creation of Kinesis streams.
gg.hander.name.shardCount	Optional	Positive integer.	1	A Kinesis stream contains one or more shards. Controls the number of shards on Kinesis streams that the Kinesis Handler creates. Multiple shards can help improve the ingest performance to a Kinesis stream. Use only when gg.handler.name.enableStreamCreation is set to true.
gg.hander.name.proxyProtocol	Optional	HTTP | HTTPS	HTTP	Sets the proxy protocol connection to the proxy server for additional level of security. The client first performs an SSL handshake with the proxy server, and then an SSL handshake with Amazon AWS. This feature was added into the Amazon SDK in version 1.11.396 so you must use at least that version to use this property.
gg.handler.name.enableSTS	Optional	true | false	false	Set to true, to enable the Kinesis Handler to access Kinesis credentials from the AWS Security Token Service. Ensure that the AWS Security Token Service is enabled if you set this property to true.
gg.handler.name.STSRegion	Optional	Any legal AWS region specifier.	The region is obtained from the gg.handler.name.region property.	Use to resolve the region for the STS call. It's only valid if the gg.handler.name.enableSTS property is set to true. You can set a different AWS region for resolving credentials from STS than the configured Kinesis region.
gg.handler.name.accessKeyId	Optional	A valid AWS access key.	None	Set this parameter to explicitly set the access key for AWS. This parameter has no effect if gg.handler.name.enableSTS is set to true. If unset, credentials resolution falls back to the AWS default credentials provider chain.
gg.handler.name.secretKey	Optional	A valid AWS secret key.	None	Set this parameter to explicitly set the secret key for AWS. This parameter has no effect if gg.handler.name.enableSTS is set to true. If unset, credentials resolution falls back to the AWS default credentials provider chain.

9.2.2.3.3 Using Templates to Resolve the Stream Name and Partition Name

The Kinesis Streams Handler provides the functionality to resolve the stream name and the partition key at runtime using a template configuration value. Templates allow you to configure static values and keywords. Keywords are used to dynamically replace the keyword with the context of the current processing. Templates are applicable to the following configuration parameters:

gg.handler.name.streamMappingTemplate
gg.handler.name.partitionMappingTemplate

Source database transactions are made up of 1 or more individual operations which are the individual inserts, updates, and deletes. The Kinesis Handler can be configured to send one message per operation (insert, update, delete, Alternatively, it can be configured to group operations into messages at the transaction level. Many of the template keywords resolve data based on the context of an individual source database operation. Therefore, many of the keywords do not work when sending messages at the transaction level. For example ${fullyQualifiedTableName} does not work when sending messages at the transaction level. The ${fullyQualifiedTableName} property resolves to the qualified source table name for an operation. Transactions can contain multiple operations for many source tables. Resolving the fully-qualified table name for messages at the transaction level is non-deterministic and so abends at runtime.

For more information about the Template Keywords, see Template Keywords.

Example Templates

The following describes example template configuration values and the resolved values.

Example Template	Resolved Value
${groupName}_${fullyQualifiedTableName}	KINESIS001_DBO.TABLE1
prefix_${schemaName}_${tableName}_suffix	prefix_DBO_TABLE1_suffix
${currentTimestamp[yyyy-mm-dd hh:MM:ss.SSS]}	2017-05-17 11:45:34.254

9.2.2.3.4 Resolving AWS Credentials

• AWS Kinesis Client Authentication
  The Kinesis Handler is a client connection to the AWS Kinesis cloud service. The AWS cloud must be able to successfully authenticate the AWS client in order in order to successfully interface with Kinesis.

9.2.2.3.4.1 AWS Kinesis Client Authentication

The Kinesis Handler is a client connection to the AWS Kinesis cloud service. The AWS cloud must be able to successfully authenticate the AWS client in order in order to successfully interface with Kinesis.

The AWS client authentication has become increasingly complicated as more authentication options have been added to the Kinesis Stream Handler. This topic explores the different use cases for AWS client authentication.

• Explicit Configuration of the Client ID and Secret
  A client ID and secret are generally the required credentials for the Kinesis Handler to interact with Amazon Kinesis. A client ID and secret are generated using the Amazon AWS website.
• Use of the AWS Default Credentials Provider Chain
  If the gg.eventhandler.name.accessKeyId and gg.eventhandler.name.secretKey are unset, then credentials resolution reverts to the AWS default credentials provider chain. The AWS default credentials provider chain provides various ways by which the AWS credentials can be resolved.
• AWS Federated Login
  The use case is when you have your on-premise system login integrated with AWS. This means that when you log into an on-premise machine, you are also logged into AWS.

9.2.2.3.4.1.1 Explicit Configuration of the Client ID and Secret

A client ID and secret are generally the required credentials for the Kinesis Handler to interact with Amazon Kinesis. A client ID and secret are generated using the Amazon AWS website.

These credentials can be explicitly configured in the Java Adapter Properties file as follows:

gg.handler.name.accessKeyId=
gg.handler.name.secretKey=

Furthermore, the Oracle Wallet functionality can be used to encrypt these credentials.

9.2.2.3.4.1.2 Use of the AWS Default Credentials Provider Chain

If the gg.eventhandler.name.accessKeyId and gg.eventhandler.name.secretKey are unset, then credentials resolution reverts to the AWS default credentials provider chain. The AWS default credentials provider chain provides various ways by which the AWS credentials can be resolved.

For more information about the default credential provider chain and order of operations for AWS credentials resolution, see Working with AWS Credentials.

When Oracle GoldenGate for Distributed Applications and Analytics (GG for DAA) runs on an AWS Elastic Compute Cloud (EC2) instance, the general use case is to resolve the credentials from the EC2 metadata service. The AWS default credentials provider chain provides resolution of credentials from the EC2 metadata service as one of the options.

9.2.2.3.4.1.3 AWS Federated Login

The use case is when you have your on-premise system login integrated with AWS. This means that when you log into an on-premise machine, you are also logged into AWS.

In this use case:

• You may not want to generate client IDs and secrets. (Some users disable this feature in the AWS portal).
• The client AWS applications need to interact with the AWS Security Token Service (STS) to obtain an authentication token for programmatic calls made to Kinesis.

This feature is enabled by setting the following: gg.eventhandler.name.enableSTS=true.

9.2.2.3.5 Configuring the Proxy Server for Kinesis Streams Handler

Oracle GoldenGate can be used with a proxy server using the following parameters to enable the proxy server:

gg.handler.name.proxyServer= 
gg.handler.name.proxyPort=80
gg.handler.name.proxyUsername=username
gg.handler.name.proxyPassword=password

Sample configurations:

gg.handlerlist=kinesis 
gg.handler.kinesis.type=kinesis_streams 
gg.handler.kinesis.mode=op 
gg.handler.kinesis.format=json 
gg.handler.kinesis.region=us-west-2 
gg.handler.kinesis.partitionMappingTemplate=TestPartitionName
gg.handler.kinesis.streamMappingTemplate=TestStream
gg.handler.kinesis.deferFlushAtTxCommit=true 
gg.handler.kinesis.deferFlushOpCount=1000 
gg.handler.kinesis.formatPerOp=true 
#gg.handler.kinesis.customMessageGrouper=oracle.goldengate.handler.kinesis.KinesisJsonTxMessageGrouper 
gg.handler.kinesis.proxyServer=www-proxy.myhost.com 
gg.handler.kinesis.proxyPort=80

9.2.2.3.6 Configuring Security in Kinesis Streams Handler

The Amazon Web Services (AWS) Kinesis Java SDK uses HTTPS to communicate with Kinesis. Mutual authentication is enabled. The AWS server passes a Certificate Authority (CA) signed certificate to the AWS client which allow the client to authenticate the server. The AWS client passes credentials (client ID and secret) to the AWS server which allows the server to authenticate the client.

9.2.2.4 Kinesis Handler Performance Considerations

• Kinesis Streams Input Limitations
  
• Transaction Batching
  
• Deferring Flush at Transaction Commit
  

9.2.2.4.1 Kinesis Streams Input Limitations

The maximum write rate to a Kinesis stream with a single shard to be 1000 messages per second up to a maximum of 1MB of data per second. You can scale input to Kinesis by adding additional Kinesis streams or adding shards to streams. Both adding streams and adding shards can linearly increase the Kinesis input capacity and thereby improve performance of the Oracle GoldenGate Kinesis Streams Handler.

Adding streams or shards can linearly increase the potential throughput such as follows:

• 1 stream with 2 shards = 2000 messages per second up to a total data size of 2MB per second.

• 3 streams of 1 shard each = 3000 messages per second up to a total data size of 3MB per second.

To fully take advantage of streams and shards, you must configure the Oracle GoldenGate Kinesis Streams Handler to distribute messages as evenly as possible across streams and shards.

Adding additional Kinesis streams or shards does nothing to scale Kinesis input if all data is sent to using a static partition key into a single Kinesis stream. Kinesis streams are resolved at runtime using the selected mapping methodology. For example, mapping the source table name as the Kinesis stream name may provide good distribution of messages across Kinesis streams if operations from the source trail file are evenly distributed across tables. Shards are selected by a hash of the partition key. Partition keys are resolved at runtime using the selected mapping methodology. Therefore, it is best to choose a mapping methodology to a partition key that rapidly changes to ensure a good distribution of messages across shards.

9.2.2.4.2 Transaction Batching

The Oracle GoldenGate Kinesis Streams Handler receives messages and then batches together messages by Kinesis stream before sending them via synchronous HTTPS calls to Kinesis. At transaction commit all outstanding messages are flushed to Kinesis. The flush call to Kinesis impacts performance. Therefore, deferring the flush call can dramatically improve performance.

The recommended way to defer the flush call is to use the GROUPTRANSOPS configuration in the replicat configuration. The GROUPTRANSOPS groups multiple small transactions into a single larger transaction deferring the transaction commit call until the larger transaction is completed. The GROUPTRANSOPS parameter works by counting the database operations (inserts, updates, and deletes) and only commits the transaction group when the number of operations equals or exceeds the GROUPTRANSOPS configuration setting. The default GROUPTRANSOPS setting for replicat is 1000.

Interim flushes to Kinesis may be required with the GROUPTRANSOPS setting set to a large amount. An individual call to send batch messages for a Kinesis stream cannot exceed 500 individual messages or 5MB. If the count of pending messages exceeds 500 messages or 5MB on a per stream basis then the Kinesis Handler is required to perform an interim flush.

9.2.2.4.3 Deferring Flush at Transaction Commit

The messages are by default flushed to Kinesis at transaction commit to ensure write durability. However, it is possible to defer the flush beyond transaction commit. This is only advisable when messages are being grouped and sent to Kinesis at the transaction level (that is one transaction = one Kinesis message or chunked into a small number of Kinesis messages), when the user is trying to capture the transaction as a single messaging unit.

This may require setting the GROUPTRANSOPS replication parameter to 1 so as not to group multiple smaller transactions from the source trail file into a larger output transaction. This can impact performance as only one or few messages are sent per transaction and then the transaction commit call is invoked which in turn triggers the flush call to Kinesis.

In order to maintain good performance the Oracle GoldenGate Kinesis Streams Handler allows the user to defer the Kinesis flush call beyond the transaction commit call. The Oracle GoldenGate replicat process maintains the checkpoint in the .cpr file in the {GoldenGate Home}/dirchk directory. The Java Adapter also maintains a checkpoint file in this directory named .cpj. The Replicat checkpoint is moved beyond the checkpoint for which the Oracle GoldenGate Kinesis Handler can guarantee message loss will not occur. However, in this mode of operation the GoldenGate Kinesis Streams Handler maintains the correct checkpoint in the .cpj file. Running in this mode will not result in message loss even with a crash as on restart the checkpoint in the .cpj file is parsed if it is before the checkpoint in the .cpr file.

9.2.2.5 Troubleshooting

• Java Classpath
  
• Kinesis Handler Connectivity Issues
  
• Logging
  

9.2.2.5.1 Java Classpath

The most common initial error is an incorrect classpath to include all the required AWS Kinesis Java SDK client libraries and creates a ClassNotFound exception in the log file.

You can troubleshoot by setting the Java Adapter logging to DEBUG, and then rerun the process. At the debug level, the logging includes information about which JARs were added to the classpath from the gg.classpath configuration variable.

The gg.classpath variable supports the wildcard asterisk (*) character to select all JARs in a configured directory. For example, /usr/kinesis/sdk/*, see Setting Up and Running the Kinesis Streams Handler.

9.2.2.5.2 Kinesis Handler Connectivity Issues

If the Kinesis Streams Handler is unable to connect to Kinesis when running on premise, the problem can be the connectivity to the public Internet is protected by a proxy server. Proxy servers act a gateway between the private network of a company and the public Internet. Contact your network administrator to get the URLs of your proxy server, and then follow the directions in Configuring the Proxy Server for Kinesis Streams Handler.

9.2.2.5.3 Logging

The Kinesis Streams Handler logs the state of its configuration to the Java log file.

This is helpful because you can review the configuration values for the handler. Following is a sample of the logging of the state of the configuration:

**** Begin Kinesis Streams Handler - Configuration Summary ****   
Mode of operation is set to op.   
   The AWS region name is set to [us-west-2].   
   A proxy server has been set to [www-proxy.us.oracle.com] using port [80].   
   The Kinesis Streams Handler will flush to Kinesis at transaction commit.  
    Messages from the GoldenGate source trail file will be sent at the operation level. 
   One operation = One Kinesis Message   
The stream mapping template of [${fullyQualifiedTableName}] resolves to [fully qualified table name].  
 The partition mapping template of [${primaryKeys}] resolves to [primary keys].   
**** End Kinesis Streams Handler - Configuration Summary ****