software.amazon.kinesis.coordinator.assignment

Class MigrationAwareLAMDataManager

java.lang.Object

software.amazon.kinesis.coordinator.assignment.MigrationAwareLAMDataManager

All Implemented Interfaces:

LAMDataManager

@ThreadSafe
public class MigrationAwareLAMDataManager
extends Object
implements LAMDataManager

Migration-aware implementation of LAMDataManager that encapsulates:

• Reading leases and worker metrics from the lease table (via EntityDAO) and optionally the legacy table (via WorkerMetricStatsDAO's legacy delegate)
• Filtering out invalid worker metrics (logging + emitting metrics)
• Filtering to only valid, non-expired worker metrics
• Async deletion of stale worker metrics (using the correct delegate based on origin)
• Computing and publishing TableMigrationSummary for migration state decisions

LAM receives valid, non-expired worker metrics and applies its own domain-specific filtering (e.g., excluding workers with failing metrics from assignment).

The main purpose of this class it to reuse the scan result from LAM for other things like tableMigrationStateMachine. Ideally the two should be decoupled and can do their own scan to get the result they want separately at the cost of higher DDB usage. To make table migration seemless keep the iops low and being efficient is better.

Field Summary

Fields

Modifier and Type	Field and Description
static int	DEFAULT_NO_OF_SKIP_STAT_FOR_DEAD_WORKER_THRESHOLD

Constructor Summary

Constructors

Constructor and Description
MigrationAwareLAMDataManager(EntityDAO entityDAO, WorkerMetricStatsDAO workerMetricsDAO, TableMigrationStatusProvider tableMigrationStatusProvider, Consumer<TableMigrationSummary> migrationSummaryConsumer, LeaseManagementConfig.WorkerUtilizationAwareAssignmentConfig config, ExecutorService executorService)

Method Summary

Modifier and Type	Method and Description
LAMDataSnapshot	loadData(MetricsScope metricsScope) Load, validate, and clean worker metrics and leases for a single LAM assignment cycle.
void	shutdown() Shuts down any internal resources (e.g., thread pools) used by this manager.

Methods inherited from class java.lang.Object

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

Field Detail

DEFAULT_NO_OF_SKIP_STAT_FOR_DEAD_WORKER_THRESHOLD

public static final int DEFAULT_NO_OF_SKIP_STAT_FOR_DEAD_WORKER_THRESHOLD

See Also:

Constant Field Values

Constructor Detail

MigrationAwareLAMDataManager

public MigrationAwareLAMDataManager(EntityDAO entityDAO,
                                    WorkerMetricStatsDAO workerMetricsDAO,
                                    TableMigrationStatusProvider tableMigrationStatusProvider,
                                    Consumer<TableMigrationSummary> migrationSummaryConsumer,
                                    LeaseManagementConfig.WorkerUtilizationAwareAssignmentConfig config,
                                    ExecutorService executorService)

Method Detail

loadData

public LAMDataSnapshot loadData(MetricsScope metricsScope)

Description copied from interface: LAMDataManager

Load, validate, and clean worker metrics and leases for a single LAM assignment cycle.

The implementation will:

1. Scan the lease table for LEASE and WORKER_METRIC_STATS entities (single DDB scan)
2. If table migration is not complete, also scan the legacy worker-metrics table
3. Filter out invalid worker metrics, log and emit metrics for them
4. Filter out expired/stale worker metrics, trigger async deletion
5. Emit count of workers with failing worker metrics
6. Compute migration summary and push them to the registered consumer
7. Return only active, valid leases and worker metrics

Specified by:

loadData in interface LAMDataManager

Parameters:

metricsScope - the metrics scope for emitting data quality metrics

Returns:

a snapshot containing clean leases and only active, valid worker metrics for LAM

shutdown

public void shutdown()

Description copied from interface: LAMDataManager

Shuts down any internal resources (e.g., thread pools) used by this manager. Should be called when the LeaseAssignmentManager is stopped.

Specified by:

shutdown in interface LAMDataManager