software.amazon.kinesis.worker.metricstats

Class WorkerMetricStatsDAO

java.lang.Object

software.amazon.kinesis.worker.metricstats.WorkerMetricStatsDAO

public class WorkerMetricStatsDAO
extends Object

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

Read Logic (getAllWorkerMetricStats):

Uses the dynamic (live) status from TableMigrationStatusProvider. If status == COMPLETE, reads only from lease table. Otherwise combines results from both legacy and lease table delegates. Caller is responsible for determining which entries are current/latest/not expired.

Write/Delete Logic:

Uses the cached status captured at initialization time so that the write target does not change dynamically during the lifetime of this DAO instance. 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 workers flip to Lease table when it reboots. This is to ensure the worker writes move as the deployment progresses in the cluster fleet.

Constructor Summary

Constructors

Constructor and Description
WorkerMetricStatsDAO(software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient dynamoDbAsyncClient, LeaseManagementConfig.WorkerMetricsTableConfig workerMetricsTableConfig, String leaseTableName, Long workerMetricsReporterFrequencyMillis, TableMigrationStatusProvider tableMigrationStatusProvider)

Method Summary

Modifier and Type	Method and Description
boolean	deleteMetrics(WorkerMetricStats workerMetrics) Delete the worker metric stats entry with a conditional check on lastUpdateTime.
List<WorkerMetricStats>	getAllWorkerMetricStats() Get all worker metric stats by combining results from both delegates.
LeaseTableWorkerMetricStatsDAODelegate	getLeaseTableDaoDelegate()
LegacyTableWorkerMetricStatsDAODelegate	getLegacyTableDaoDelegate()
void	initialize() Initialize the DAO: initializes both delegates and caches the current migration status to determine the write target for the lifetime of this DAO instance.
void	updateMetrics(WorkerMetricStats workerMetrics) Update (or create) the worker metric stats for a given worker.

Methods inherited from class java.lang.Object

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

Constructor Detail

WorkerMetricStatsDAO

public WorkerMetricStatsDAO(software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient dynamoDbAsyncClient,
                            LeaseManagementConfig.WorkerMetricsTableConfig workerMetricsTableConfig,
                            String leaseTableName,
                            Long workerMetricsReporterFrequencyMillis,
                            TableMigrationStatusProvider tableMigrationStatusProvider)

Method Detail

getLegacyTableDaoDelegate

public LegacyTableWorkerMetricStatsDAODelegate getLegacyTableDaoDelegate()

getLeaseTableDaoDelegate

public LeaseTableWorkerMetricStatsDAODelegate getLeaseTableDaoDelegate()

initialize

public void initialize()
                throws DependencyException

Initialize the DAO: initializes both delegates and caches the current migration status to determine the write target for the lifetime of this DAO instance.

Throws:

DependencyException - if unable to determine legacy table existence

InvalidStateException - if the TableMigrationStatusProvider is still UNKNOWN

getAllWorkerMetricStats

public List<WorkerMetricStats> getAllWorkerMetricStats()
                                                throws DependencyException,
                                                       InvalidStateException,
                                                       ProvisionedThroughputException

Get all worker metric stats by combining results from both delegates. If migration is COMPLETE, reads only from lease table. The caller is responsible for determining which entries are current/latest/not expired.

Returns:

list of all worker metric stats from both tables (may contain duplicates by workerId)

Throws:

DependencyException - if DDB fails unexpectedly

InvalidStateException - if a required table does not exist

ProvisionedThroughputException - if DDB lacks capacity

updateMetrics

public void updateMetrics(WorkerMetricStats workerMetrics)
                   throws DependencyException,
                          InvalidStateException,
                          ProvisionedThroughputException

Update (or create) the worker metric stats for a given worker. Routes to the appropriate table based on the cached migration status.

Parameters:

workerMetrics - the worker metrics to persist

Throws:

DependencyException - if DDB fails unexpectedly

InvalidStateException - if not initialized or table does not exist

ProvisionedThroughputException - if DDB lacks capacity

deleteMetrics

public boolean deleteMetrics(WorkerMetricStats workerMetrics)
                      throws DependencyException,
                             InvalidStateException,
                             ProvisionedThroughputException

Delete the worker metric stats entry with a conditional check on lastUpdateTime.

Parameters:

workerMetrics - the entry to delete (workerId and lastUpdateTime required)

Returns:

true if deleted, false if the conditional check failed

Throws:

DependencyException - if DDB fails unexpectedly

InvalidStateException - if not initialized or table does not exist

ProvisionedThroughputException - if DDB lacks capacity