🔥 Jadwal & Hasil: awslabs/amazon kinesis client/pull/

akidambisrinivasan · 2026-06-11T20:15:18Z

Issue #, if available:

Description of changes:

Introduce a mechanism for KCL to migrate to single lease table for all entities

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

lucienlu-aws · 2026-06-11T20:52:12Z

+        addTableMigrationStateHistoryEntry(this.tableMigrationStatus, this.modifiedBy, this.modifiedTimestamp);
+        this.tableMigrationStatus = tableMigrationStatus;
+        this.modifiedBy = modifiedBy;
+        this.modifiedTimestamp = System.currentTimeMillis();


Do any of our tests rely on this modifiedTimestamp? i.e. comparing time since this timestamp. If so, we should pass a Clock instance to the constructor to allow mocking in unit tests to avoid flaky tests

yes this is what is used for the bake time, I can pass clock to TableMigrationStateMachine.

lucienlu-aws · 2026-06-11T21:21:30Z

+        /**
+         * Get the parent {@link EntityType} for this coordinator state type.
+         */
+        public EntityType getEntityType() {


lombok @Getter?

lucienlu-aws · 2026-06-11T21:23:47Z

+    /**
+     * The string value persisted in DynamoDB as the entityType attribute.
+     */
+    public String getDdbValue() {


lombok @Getter?

Introduce EntityType enum as foundational abstraction for co-locating multiple entity types (Lease, WorkerMetricStats, CoordinatorState) in a single DynamoDB table. - Add EntityType enum with DDB-persisted values and CoordinatorStateType subset - Lease, CoordinatorState, WorkerMetricStats implement entity type contract - DynamoDBLeaseSerializer writes entityType attribute, skips non-lease on read - DynamoDBLeaseRefresher handles null returns from fromDynamoRecord gracefully - WorkerMetricStats adds Features enum, supportCode, LegacyWorkerMetricStats and LeaseTableWorkerMetricStats subclasses for different PK strategies - Add TableMigrationState and TableMigrationStatus for phased migration tracking - Add CoordinatorConfig changes to support table migration configuration - Add LeaderLock CoordinatorState subclass and DynamoDBLockBasedLeaderDecider changes - StreamInfoDAO adjustment for new CoordinatorState serialize API Note: This CR will not compile standalone — it requires all 5 changes in the TableMigration series. The series is separated to focus on a single concern per change for easier review and validation.

…actoring Introduce TableMigrationStatusProvider to control which DDB table(s) each DAO reads from and writes to during the phased table migration. - Add TableMigrationStatusProvider (thread-safe, dynamically updatable) - Add EntityDAO interface for unified multi-entity scan - Add DynamoDBLeaseTableDao as EntityDAO implementation - Add CoordinatorStateDAODelegate interface + Legacy/LeaseTable implementations - Add WorkerMetricStatsDAODelegate interface + Legacy/LeaseTable implementations - Refactor CoordinatorStateDAO to route via delegates based on migration status - Refactor WorkerMetricStatsDAO to route via delegates based on migration status Note: This CR will not compile standalone — it requires all 5 changes in the TableMigration series. The series is separated to focus on a single concern per change for easier review and validation.

Extract data loading, validation, and cleanup from LeaseAssignmentManager into a dedicated LAMDataManager interface. LAM becomes agnostic to table migration. - Add LAMDataManager interface for data loading abstraction - Add LAMDataSnapshot value object for clean data contract - Add MigrationAwareLAMDataManager: merges data from lease + legacy tables, handles worker metric validation/expiry/cleanup, computes TableMigrationSummary - Add TableMigrationSummary: stats pushed to table migration state machine - LeaseAssignmentManager: remove direct DDB loading, accept LAMDataManager, remove stale worker metrics cleanup (now in MigrationAwareLAMDataManager) Note: This CR will not compile standalone — it requires all 5 changes in the TableMigration series. The series is separated to focus on a single concern per change for easier review and validation.

Implement the 2-phase table migration state machine for consolidating CoordinatorState and WorkerMetricStats from legacy tables into the lease table. - Add TableMigrationStateMachine interface - Add TableMigrationStateMachineImpl: orchestrates INIT -> DEPLOYED -> PENDING -> COMPLETE transitions with bake time, rollback detection, and leader coordination - Wire TableMigrationStateMachine into DynamoDBLockBasedLeaderDecider (handleLeaderLockResult) - Wire TableMigrationStateMachine into Scheduler's leader decider creation - DynamoDBLockBasedLeaderDecider catches DependencyException/InvalidStateException, releases lock and returns false on failure Note: This CR will not compile standalone — it requires all 5 changes in the TableMigration series. The series is separated to focus on a single concern per change for easier review and validation.

Wire all table migration and LAM components into the Scheduler and update the client version migration flow with a new Phase 1 (passive 2.x-compatible) state. - MigrationClientVersionInitState: Phase 1 passive state (no LAM, no metrics) - DynamicMigrationComponentsInitializer: add initializeClientVersionForPhase1() - MigrationClientVersionStateInitializer: handle PHASE1 config - MigrationStateMachineImpl: integrate new init state - Scheduler: wire TableMigrationStateMachine, TableMigrationStatusProvider, DynamoDBLeaseTableDao, MigrationAwareLAMDataManager into LAM creation

Issues: 1. 3.5 migration does not read 3.4 worker metrics causing lease rebalancing to only 3.5 workers. Root cause: entityType filter on lease table scan was filtering out 3.4 worker metrics (written without entityType attribute). Fix: Legacy delegate does unfiltered scan; lease table delegate filters by entityType since it shares the table with other entity types. 2. TableMigrationState written to lease table during phase 1 of 2→3 migration when workers may not be ready to write to lease table yet. Fix: Do not persist state during 2→3 short-circuit. Can be improved later to write during phase 3 of 2→3 migration. 3. TableMigrationState did not include entityType in serialization causing deserialization errors (misinterpreted as a lease record). Fix: TableMigrationState.serialize() now calls super.serialize() to include entityType. DAO delegate strips entityType before passing to deserializers. 4. initializeClientVersionForPhase1 did not initialize leaseAssignmentModeProvider causing DynamoDbLeaseCoordinator exceptions (provider not initialized). Fix: Call leaseModeChangeConsumer.initialize() in phase 1 initialization. Other fixes: - CoordinatorStateDAODelegate: use remove() for entityType so deserializers don't see unknown attributes - StreamInfo.deserialize(): remove redundant entityType check (stripped by DAO) - WorkerMetricStatsDAODelegate: accept Logger from child class - LegacyTableWorkerMetricStatsDAODelegate: return empty list when disabled - Updated tests for all changes

…king Issues fixed: 1. isLegacyWorkerMetricsEmpty used workersWithUnexpiredLeases == totalActiveWorkersWithMetrics which fails when there are more workers emitting metrics than workers owning leases (e.g. new workers not yet assigned leases). Fix: compare leaseOwnersWithActiveMetrics against workersWithUnexpiredLeases — checks that every lease owner is emitting metrics. 2. Added totalWorkersWithLeases and leaseOwnersWithActiveMetrics to TableMigrationSummary to support the corrected readiness check. 3. 2→3 migration phase 1 workers not taking leases. DynamoDBLeaseCoordinator only started the LeaseTaker when dynamicModeChangeSupportNeeded() was true, but phase 1 sets mode to DEFAULT_LEASE_COUNT_BASED_ASSIGNMENT without dynamic mode support. Fix: also start the taker when current mode is DEFAULT_LEASE_COUNT_BASED_ASSIGNMENT. 4. NPE on shutdown path: DynamicMigrationComponentsInitializer.shutdown() called leaseAssignmentManager.stop() unconditionally, but LAM is null in phase 1. Fix: null-check before calling stop(). Refactoring: - DynamicMigrationComponentsInitializer: extract initializeStartupComponents() to centralize leader-decider wiring, optional LAM creation, workerMetrics startup, and mode-provider initialization. Eliminates 4 duplicated initialization blocks. Review comment changes: - LAMDataManager.loadData() no longer throws checked exceptions (wraps in CompletionException) - LeaseAssignmentManager: handle InterruptedException during assignment gracefully - TableMigrationStateMachineImpl: accept dedicated ExecutorService for async copy (no longer uses ForkJoinPool.commonPool()), add shutdown() lifecycle method - EntityDAO/DynamoDBLeaseTableDao: add shutdown() for thread pool cleanup, interrupt check in scan loop - Scheduler: create EntityDAO earlier, wire shutdown in finalShutdown() - CoordinatorStateDAO: handle InterruptedException separately from ExecutionException - DynamoDBLeaseSerializer: downgrade non-lease entity log from warn to debug - MigrationAwareLAMDataManager: add shutdown(), handle InterruptedException in retry loop - Suppress pre-existing unused dependency warnings in pom.xml - Added Apache license headers to new files - LAM integration test: verify assignment reaches 3.4 workers during migration - MigrationAwareLAMDataManagerTest: tests for new summary fields

@requiredargsconstructor

…re on 2x→3x_with_rollback transition Issue: When the MigrationStateMachine transitions from UPGRADE_FROM_2X to 3X_WITH_ROLLBACK, it performs a conditional update on the MigrationState in DDB expecting the current state to match. However, the initial state was never written to DDB on first enter, so the conditional expression failed (item did not exist) and the transition retried indefinitely. Fix: Extract AbstractMigrationClientVersionState base class that ensures the initial MigrationState is persisted to DDB on first enter All state classes now extend this base which handles: - Idempotent enter/leave guards (entered/left flags) - Writing initial state to DDB if it doesn't exist (fixes the conditional write precondition for subsequent transitions) Refactoring: - Removed @requiredargsconstructor from state classes in favor of explicit constructors that pass MigrationState and workerIdentifier to the base class - MigrationStateMachineImpl: pass migrationState and workerId to all state constructors - MigrationClientVersionUpgradeFrom2xState: use curentMigrationState from base class (field rename from currentMigrationState to avoid shadowing) - Add the tests which was missed in the previous commit

…der releaseLeadershipIfHeld Fix 1: Move LAMDataManager executor shutdown to application shutdown Previously, LAMDataManager.shutdown() was called inside LeaseAssignmentManager.stop(). During migration rollback and roll-forward scenarios, LAM is stopped and restarted dynamically. Shutting down the data manager's executor on LAM stop meant that when LAM was restarted, the data manager could no longer load data (its scheduler was already terminated). Move the shutdown to DynamicMigrationComponentsInitializer.shutdown() which is only called when the application is going down, not during dynamic LAM stop/start cycles. Fix 2: MigrationAdaptiveLeaderDecider.releaseLeadershipIfHeld() was a no-op When LAM fails consistently, it releases leadership via releaseLeadershipIfHeld() to allow another worker to become leader. However, MigrationAdaptiveLeaderDecider did not override this method from the LeaderDecider interface, so the default no-op implementation was used. This meant leadership was never actually released on persistent LAM failures. Add the override to delegate to the current leader decider.

…esign Issue: In single-table mode, the lease table contains non-lease entities (WorkerMetricStats, CoordinatorState) alongside lease records. The previous implementation used DynamoDB's scan 'limit' parameter to control page size, combined with a 'maxPages' parameter to cap the number of paginated scan calls. This approach was broken for single-table because: 1. DynamoDB's scan 'limit' controls items *evaluated* per request, not items *returned* after filtering. Non-lease entities count toward the limit but are deserialized as null and discarded. So a scan with limit=1 on a table where the first item is a WorkerMetricStats record would return 0 leases and report no more pages, even if leases exist later in the table. 2. isLeaseTableEmpty() called list(1, 1, null) - limit=1, maxPages=1. If the single scanned item was a non-lease entity, it would incorrectly return true (table appears empty) even though leases exist. This could cause the application to re-bootstrap shard leases or behave as if starting fresh. 3. The maxPages cap was an arbitrary safeguard that could prematurely stop pagination before all requested leases were found. Fix: - Remove the DynamoDB scan 'limit' parameter entirely. Let DynamoDB return its default page size (1MB) per request, which is more efficient for throughput. - Remove the maxPages parameter and the private 3-arg list() overload. - Implement limit as an application-level cap: iterate through scan results, collect deserialized leases, and stop when the requested count is reached or all pages are exhausted. - Use a do-while loop that continues paginating (via lastEvaluatedKey) until either the limit is satisfied or there are no more pages. - isLeaseTableEmpty() now correctly paginates past non-lease entities to determine if any actual lease exists in the table.

akidambisrinivasan · 2026-06-18T18:17:32Z

+            if (!legacyDao.isEnabled()) {
+                // No legacy table — 2→3 migration short-circuit: write COMPLETE to lease table directly.
+                log.info("No state in DDB, no legacy table -> writing COMPLETE (2->3 migration)");
+                writeStatusToLeaseTable(TableMigrationStatus.TABLE_MIGRATION_STATUS_COMPLETE, null);


note: this was removed in later commits so please ignore it

…nd threadpool shutdown Fix 1: TableMigrationStateMachineImpl - handleInitState reconciliation Issue: When a leader successfully writes DEPLOYED status to DDB but the local provider update fails (e.g., due to a timeout or perceived failure), the local provider remains at INIT while DDB is at DEPLOYED. On the next iteration, handleInitState reads DDB and finds DEPLOYED, but the previous code would incorrectly attempt to use the DEPLOYED record's modifiedTimestamp as the INIT bake start time, computing a nonsensical elapsed duration. This could either stall the state machine (if the timestamp was recent) or cause premature transitions. Fix: When handleInitState reads DDB and finds the status is DEPLOYED (not the expected INIT), reconcile immediately by updating the local provider to DEPLOYED. This causes the next iteration to dispatch to handleDeployedState, which is the correct handler. Also add a guard for other unexpected DDB states (PENDING, COMPLETE) which should not occur when local is INIT but are logged defensively. Fix 2: WorkerMetricStatsDAO - dynamic read routing on COMPLETE Issue: The read path (getAllWorkerMetricStats) used the cached status captured at initialization time. This meant that after the migration transitions to COMPLETE, workers would continue reading from both legacy and lease tables until they restarted. This is wasteful and could surface stale/expired entries from the legacy table that are no longer being written. Fix: Change getAllWorkerMetricStats to use the live status from TableMigrationStatusProvider instead of the cached status. When status becomes COMPLETE dynamically, reads immediately flip to lease-table-only. Write routing still uses the cached status (updated only on restart) to ensure writes move with the deployment wave across the fleet, not all at once. The asymmetry is intentional: - Reads: safe to dynamically flip because both tables may have data, and once COMPLETE is reached no new data is written to legacy. Reading only lease table avoids surfacing stale legacy entries. - Writes: must stay on the cached status so that each worker transitions its write target as it restarts during deployment, preventing a thundering herd of all workers simultaneously writing to the lease table. Fix 3: TableMigrationStatus - use 0L instead of null for bake times Changed bake time defaults from null to 0L for UNKNOWN, INIT, and PENDING states. Using null required null-checks throughout and risked NPEs. 0L is a safer default that means no bake time required for these states. Fix 4: DynamicMigrationComponentsInitializer - threadpool shutdown bug The shutdownThreadPool helper method was incorrectly calling shutdownNow() on lamThreadPool instead of the threadPool parameter passed to the method. This meant that when shutting down the workerMetricsScheduler, it would accidentally force- shutdown the LAM thread pool instead.

Keep javadoc consistent across files based on the actual implementation

Fix java docs to be consistent. Fix the issue where lock client routing was not dynamic, it was only done at startup time, but dynamically we need the lock client to switch once status changes to COMPLETE

Since KCL no longer support separate tables the corresponding fields are marked as deprecated

akidambisrinivasan had a problem deploying to manual-approval June 11, 2026 20:15 — with GitHub Actions Failure

lucienlu-aws reviewed Jun 11, 2026

View reviewed changes

akidambisrinivasan had a problem deploying to manual-approval June 12, 2026 16:01 — with GitHub Actions Failure

nikhil-m-amazon reviewed Jun 12, 2026

View reviewed changes

akidambisrinivasan had a problem deploying to manual-approval June 15, 2026 18:37 — with GitHub Actions Failure

akidambisrinivasan had a problem deploying to manual-approval June 16, 2026 13:45 — with GitHub Actions Failure

akidambisrinivasan had a problem deploying to manual-approval June 16, 2026 22:10 — with GitHub Actions Failure

akidambisrinivasan had a problem deploying to manual-approval June 17, 2026 00:53 — with GitHub Actions Failure

akidambisrinivasan added 10 commits June 17, 2026 05:57

Use consistentRead(true) for CoordinatorStateDAODelegate scan

3a7e6d2

akidambisrinivasan force-pushed the single_table branch from c7c1855 to 1c5d299 Compare June 17, 2026 13:19

akidambisrinivasan had a problem deploying to manual-approval June 17, 2026 13:19 — with GitHub Actions Failure

Fix compilation from rebase to TOT

2f229c8

akidambisrinivasan had a problem deploying to manual-approval June 17, 2026 17:11 — with GitHub Actions Failure

akidambisrinivasan temporarily deployed to manual-approval June 17, 2026 19:52 — with GitHub Actions Inactive