software.amazon.kinesis.coordinator

Class CoordinatorStateDAO

java.lang.Object

software.amazon.kinesis.coordinator.CoordinatorStateDAO

@ThreadSafe
public class CoordinatorStateDAO
extends Object

Data Access Object that routes CoordinatorState operations to the appropriate DDB table based on the current TableMigrationStatus.

Read Logic

• If status != COMPLETE: read from legacy table first; if found, return it. Otherwise read from lease table. The legacy delegate handles the case where the table doesn't exist (returns null/empty).
• If status == COMPLETE: read only from lease table.

Write Logic

• cache table migration startup upon initialize().
• If cachedStatus == COMPLETE or PENDING: write to lease table only.
• If cachedStatus == INIT or DEPLOYED: write to legacy table only.

The difference between read and write are: 1. Reads are merged from both tables while writes go only to one table. This is because when status is INIT/DEPLOYED, the application could be on a rollback + rollforward path, so even if application is currently writing to Legacy table, in the past workers may have written to Lease table. So its safer to read both until the final transition to COMPLETED. 2. Reads dynamically flip on completed state, but wokers flip to Lease table when it reboots. This is to ensure the worker writes move as the deployment progresses in the cluster fleet.

Special Cases

• The table migration state machine directly uses the leaseTableDaoDelegate field for its reads/writes and sets the TableMigrationStatusProvider before calling initialize().
• All operations require initialize() to have been called first.

Lock Client

The lock client options builder routes based on status:

• COMPLETE: lease table lock
• All others (INIT, DEPLOYED, PENDING, UNKNOWN): legacy table lock

During transition from PENDING to COMPLETE, its possible workers that are caught up to COMPLETE state will grab lock from lease table however those that grab the lock when their local state is still PENDING will grab from the legacy lock. However TableMigrationStateMachine guarantees that it performs a post leadership acquire check where if the local state is different from the DDB state it will force the lock to be relinquished and then return false to is caller on isLeader method. The next isLeader call will use the COMPLETED status to grab the lock correctly. Given that you can never rollback from COMPLETED means that we dont have a rollback scenario to handle correct lock acquision that guarantees only a single leader at a time. Refer to TableMigrationStateMachine.handleLeaderLockResult

TODO

Update and delete operations currently route based on the table migration status which may not match where the entry was originally read from. If a caller reads from legacy but the status transitions to PENDING/COMPLETE before the update, the update targets the lease table where the entry may not yet exist. Consider tracking read-source per entry and routing to the appropriate delegate based on the source. For now, the routing is best-effort and callers that perform read-then-update must handle conditional check failures gracefully. There is no usecase for an update coordinator state except for the migration state and table migration state. Table Migration state is always updated from state machine using appropriate delegate class directly and never through this DOA. For more details refer to TableMigrationStateMachine. And for migration state the assumption is both state machines are not running concurrently.

Constructor Summary

Constructors

Constructor and Description
CoordinatorStateDAO(software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient dynamoDbAsyncClient, CoordinatorConfig.CoordinatorStateTableConfig coordinatorStateTableConfig, String leaseTableName, TableMigrationStatusProvider tableMigrationStatusProvider)

Method Summary

Modifier and Type	Method and Description
boolean	createCoordinatorStateIfNotExists(@NonNull CoordinatorState state) Create a coordinator state if it does not already exist.
boolean	deleteCoordinatorState(@NonNull String key) Delete a coordinator state by key.
void	executeTransactWrite(List<software.amazon.awssdk.services.dynamodb.model.TransactWriteItem> transactWriteItems) Execute a DynamoDB TransactWriteItems request with the given list of TransactWriteItems.
CoordinatorState	getCoordinatorState(@NonNull String key) Get a single CoordinatorState by key.
com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient	getDDBLockClient() Get the active DDB lock client based on the current table migration status.
LeaseTableCoordinatorStateDAODelegate	getLeaseTableDaoDelegate()
LegacyTableCoordinatorStateDAODelegate	getLegacyTableDaoDelegate()
void	initialize() Initialize the DAO for write operations.
void	initializeDelegates() Initialize the delegates for read operations.
void	initializeLockClients(long leaseDurationMillis, long heartbeatPeriodMillis, String workerId) Initialize the DDB lock clients for leader election.
List<CoordinatorState>	listCoordinatorState() List all coordinator states.
List<CoordinatorState>	listCoordinatorStateByEntityType(EntityType.CoordinatorStateType entityType) List coordinator states by entity type.
void	shutdownLockClients() Shutdown both lock clients, closing their background heartbeat threads.
boolean	updateCoordinatorStateWithExpectation(@NonNull CoordinatorState state, Map<String,software.amazon.awssdk.services.dynamodb.model.ExpectedAttributeValue> expectations) Update a coordinator state with expectations.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

CoordinatorStateDAO

public CoordinatorStateDAO(software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient dynamoDbAsyncClient,
                           CoordinatorConfig.CoordinatorStateTableConfig coordinatorStateTableConfig,
                           String leaseTableName,
                           TableMigrationStatusProvider tableMigrationStatusProvider)

Method Detail

getLegacyTableDaoDelegate

public LegacyTableCoordinatorStateDAODelegate getLegacyTableDaoDelegate()

getLeaseTableDaoDelegate

public LeaseTableCoordinatorStateDAODelegate getLeaseTableDaoDelegate()

initializeDelegates

public void initializeDelegates()
                         throws DependencyException

Initialize the delegates for read operations. This must be called early in startup (before the TableMigrationStatusProvider is initialized) so that reads can work during the table migration state machine bootstrap. The legacy delegate checks if the table exists and disables itself if not. This breaks the circular dependency: delegates can serve reads before the TableMigrationStatusProvider knows the final status.

Throws:

DependencyException - if unable to determine legacy table existence

initialize

public void initialize()
                throws InvalidStateException

Initialize the DAO for write operations. Must be called after initializeDelegates() and after the TableMigrationStatusProvider has moved past UNKNOWN.

Throws:

InvalidStateException - if the TableMigrationStatusProvider is still UNKNOWN

getCoordinatorState

public CoordinatorState getCoordinatorState(@NonNull
                                            @NonNull String key)
                                     throws DependencyException,
                                            InvalidStateException,
                                            ProvisionedThroughputException

Get a single CoordinatorState by key. Uses the fallback read pattern: if status != COMPLETE, tries legacy first then lease table. If status == COMPLETE, reads only from lease table.

Parameters:

key - the coordinator state key

Returns:

the state, or null if not found in any table

Throws:

DependencyException - if DDB fails unexpectedly

InvalidStateException - if a required table does not exist

ProvisionedThroughputException - if DDB lacks capacity

listCoordinatorState

public List<CoordinatorState> listCoordinatorState()
                                            throws ProvisionedThroughputException,
                                                   DependencyException,
                                                   InvalidStateException

List all coordinator states. Uses the fallback read pattern with merge: if status != COMPLETE, reads from both tables. Based on the write implementation there should not be any duplicate, if there is, it should be identical and all copies are returned to the caller without deduplication.

Returns:

list of all coordinator states

Throws:

DependencyException - if DDB fails unexpectedly

InvalidStateException - if a required table does not exist

ProvisionedThroughputException - if DDB lacks capacity

listCoordinatorStateByEntityType

public List<CoordinatorState> listCoordinatorStateByEntityType(EntityType.CoordinatorStateType entityType)
                                                        throws ProvisionedThroughputException,
                                                               DependencyException,
                                                               InvalidStateException

List coordinator states by entity type.

Parameters:

entityType - the entity type to filter by

Returns:

list of matching coordinator states for the given entityType

Throws:

DependencyException - if DDB fails unexpectedly

InvalidStateException - if a required table does not exist

ProvisionedThroughputException - if DDB lacks capacity

createCoordinatorStateIfNotExists

public boolean createCoordinatorStateIfNotExists(@NonNull
                                                 @NonNull CoordinatorState state)
                                          throws DependencyException,
                                                 InvalidStateException,
                                                 ProvisionedThroughputException

Create a coordinator state if it does not already exist.

Parameters:

state - the state to create

Returns:

true if created, false if the key already exists

Throws:

DependencyException - if DDB fails unexpectedly

InvalidStateException - if not initialized or table does not exist

ProvisionedThroughputException - if DDB lacks capacity

updateCoordinatorStateWithExpectation

public boolean updateCoordinatorStateWithExpectation(@NonNull
                                                     @NonNull CoordinatorState state,
                                                     Map<String,software.amazon.awssdk.services.dynamodb.model.ExpectedAttributeValue> expectations)
                                              throws DependencyException,
                                                     InvalidStateException,
                                                     ProvisionedThroughputException

Update a coordinator state with expectations.

Parameters:

state - the state to update

expectations - conditional expectations for the update

Returns:

true if updated, false if expectations not met

Throws:

DependencyException - if DDB fails unexpectedly

InvalidStateException - if not initialized or table does not exist

ProvisionedThroughputException - if DDB lacks capacity

deleteCoordinatorState

public boolean deleteCoordinatorState(@NonNull
                                      @NonNull String key)
                               throws ProvisionedThroughputException,
                                      InvalidStateException,
                                      DependencyException

Delete a coordinator state by key.

Parameters:

key - the key to delete

Returns:

true if deleted

Throws:

DependencyException - if DDB fails unexpectedly

InvalidStateException - if not initialized or table does not exist

ProvisionedThroughputException - if DDB lacks capacity

initializeLockClients

public void initializeLockClients(long leaseDurationMillis,
                                  long heartbeatPeriodMillis,
                                  String workerId)

Initialize the DDB lock clients for leader election. Creates lock clients for both the legacy table and the lease table, unless migration is already COMPLETE (in which case only the lease table lock client is created).

The lock clients have background heartbeat threads that keep acquired locks alive. An idle lock client (one that hasn't acquired a lock) will have a running heartbeat thread but it will be effectively a no-op since there are no locks to heartbeat.

This method is called once during startup and is not thread-safe with respect to concurrent calls. It must be called before any calls to getDDBLockClient().

Parameters:

leaseDurationMillis - the lock lease duration in milliseconds

heartbeatPeriodMillis - the heartbeat period in milliseconds

workerId - the owner name for the lock

getDDBLockClient

public com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient getDDBLockClient()

Get the active DDB lock client based on the current table migration status. Routes to the lease table lock client when migration status is COMPLETE, otherwise routes to the legacy table lock client.

Thread-safe: the fields are assigned once during initializeLockClients(long, long, java.lang.String) (startup) and never reassigned. The routing decision is based on the TableMigrationStatusProvider which is updated atomically. The caller (isLeader()) is synchronized, providing happens-before guarantees.

Returns:

the appropriate AmazonDynamoDBLockClient for the current migration state

shutdownLockClients

public void shutdownLockClients()

Shutdown both lock clients, closing their background heartbeat threads. Should be called during scheduler shutdown to ensure no background threads remain running.

executeTransactWrite

public void executeTransactWrite(List<software.amazon.awssdk.services.dynamodb.model.TransactWriteItem> transactWriteItems)
                          throws DependencyException

Execute a DynamoDB TransactWriteItems request with the given list of TransactWriteItems. Used by the table migration state machine to atomically move entries between tables (put into lease table + delete from legacy table in a single transaction).

Parameters:

transactWriteItems - the list of transact write items (max 100 per DDB limit)

Throws:

DependencyException - if DDB fails unexpectedly or the transaction is cancelled