Add support for multi-tenancy for lease coordination and management in KCL

[sb2k16]

Problem Description

Currently, whenever a new KCL application is launched, it automatically creates a DynamoDB table for lease coordination and management for the specific application.

It would be nice to have KCL provide MultiTenant support such that multiple independent KCL applications (hosted in a single account) each consuming multiple Kinesis data streams be able to share a single DynamoDB table for lease coordination and management.

Proposed Solution
The proposal is to have KCL support multi-tenancy for lease coordination and management through a single DynamoDB table with the relevant partition keys, range keys and indexes defined as required. Moreover, clients should also be offered to provide additional properties like SSE, TTL, etc.

Following Table schema could be used to support multi-tenancy.
Table partition details:

• Partition Key -
  • applicationName - Uniquely identifies the application.
• Sort Key
  • LeaseKey - To allow for get/updates of the specific lease.

Additionally, introduce a Global Secondary Index (GSI) to allow for listing all shards per stream per application.

• Partition Key -
  • applicationName - Uniquely identifies the pipeline
• Sort Key
  • streamName - Uniquely identifies a stream.

New classes and overrides

• MultiTenantDynamoDBLeaseRefresher extends DynamoDBLeaseRefresher
  • Methods:
    • public boolean createLeaseTableIfNotExists() - to support table creation if it does not exist with the required GSI, enable encryption, point-in-time-recovery.
    • private List list(Integer limit, Integer maxPages, StreamIdentifier streamIdentifier) - use dynamoDb query on the GSI to list all the shards for a stream and application.
• MultiTenantDynamoDBLeaseSerializer extends DynamoDBMultiStreamLeaseSerializer
  • Methods:
    • public Map<String, AttributeValue> toDynamoRecord(final Lease lease) - to support serialization of MultiTenantLease for insertion in the lease table
    • public Lease fromDynamoRecord(final Map<String, AttributeValue> dynamoRecord) - to support deserialization of the record from the lease table into a MultiTenantLease
    • public Map<String, AttributeValue> getDynamoHashKey(final Lease lease) - defines the hash key or partition key which is the application name and lease key as the sort key
    • public Collection getKeySchema() - defines the schema of the partition key and sort keys
  • New method
    • public Collection getIndexSchema() - defines the schema for the partition key (application name) and the sort key as the stream id
• MultiTenantDynamoDBLeaseManagementFactory extends DynamoDBLeaseManagementFactory
  • Methods:
    • public DynamoDBLeaseRefresher createLeaseRefresher() - to create the Multitenant DynamoDB lease refresher.
• LeaseSerializer interface
  • New Method
    • Collection getIndexSchema();
• MultiTenantHierarchicalShardSyncer extends HierarchicalShardSyncer
  • Methods:
    • public synchronized Lease createLeaseForChildShard - create a MultiTenantLease from a child shard
• MultiTenantLeaseManagementConfig extends LeaseManagementConfig - lease management configuration to support MultiTenant lease management functionality
• MultiTenantMultiStreamConfigsBuilder extends ConfigsBuilder
  • Methods:
    • public LeaseManagementConfig leaseManagementConfig() - return the MultiTenantLeaseManagementConfig object
• MultiTenantLease - store lease object model