Class PartitionMigrationPipeline

java.lang.Object

com.loomcache.server.cluster.PartitionMigrationPipeline

public class PartitionMigrationPipeline
extends Object

Partition migration pipeline for data redistribution during cluster topology changes.

Performs real data movement through five stages:

• PREPARE: Lock the partition, mark it as migrating, capture source data snapshot
• COPY: Copy all entries belonging to this partition from source maps to target buffer
• VERIFY: Compare CRC32 checksums and entry counts between source snapshot and copied data
• COMMIT: Update partition ownership in the PartitionTable
• CLEANUP: Remove migrated entries from the source node's maps

Thread safety: Uses ReentrantLock for pipeline coordination, per-partition locks for migration isolation, and ConcurrentHashMap for concurrent progress tracking.

Since:

1.0

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	PartitionMigrationPipeline.MigratedEntry	Value plus absolute expiry metadata for a migrated map entry.
static interface	PartitionMigrationPipeline.MigrationDataSender	Sends partition data to the target node.
static final record	PartitionMigrationPipeline.MigrationProgress	Progress record for tracking migration advancement.
static enum	PartitionMigrationPipeline.MigrationState	Migration state enum representing the lifecycle of a migration task.
static final record	PartitionMigrationPipeline.MigrationStats	Statistics record for migration pipeline metrics.
static final record	PartitionMigrationPipeline.MigrationTask	Task record representing a single partition migration.
static interface	PartitionMigrationPipeline.OwnershipCommitter	Commits a partition ownership change through Raft consensus.

Field Summary

Fields

Modifier and Type	Field	Description
static final int	DEFAULT_CONCURRENT_LIMIT
static final int	DEFAULT_MAX_RETRIES

Constructor Summary

Constructors

Constructor	Description
PartitionMigrationPipeline(int instanceNumber, DataStructureRegistry registry, PartitionTable partitionTable)
PartitionMigrationPipeline(int instanceNumber, DataStructureRegistry registry, PartitionTable partitionTable, int maxConcurrent, int maxRetries)
PartitionMigrationPipeline(int instanceNumber, DataStructureRegistry registry, PartitionTable partitionTable, @Nullable PartitionMigrationPipeline.MigrationDataSender dataSender, @Nullable PartitionMigrationPipeline.OwnershipCommitter ownershipCommitter, int maxConcurrent, int maxRetries)
PartitionMigrationPipeline(int instanceNumber, DataStructureRegistry defaultRegistry, IntFunction<DataStructureRegistry> registryForSlot, PartitionTable partitionTable, @Nullable PartitionMigrationPipeline.MigrationDataSender dataSender, @Nullable PartitionMigrationPipeline.OwnershipCommitter ownershipCommitter, int maxConcurrent, int maxRetries)	Session 10 ESC-12.3: sharded-aware constructor for the historical one-registry-per-slot mode.
PartitionMigrationPipeline(int instanceNumber, DataStructureRegistry defaultRegistry, IntFunction<DataStructureRegistry> registryForSlot, IntFunction<Collection<DataStructureRegistry>> registriesForSlot, Function<String, DataStructureRegistry> registryForKey, PartitionTable partitionTable, @Nullable PartitionMigrationPipeline.MigrationDataSender dataSender, @Nullable PartitionMigrationPipeline.OwnershipCommitter ownershipCommitter, int maxConcurrent, int maxRetries)

Method Summary

Modifier and Type	Method	Description
boolean	cancelMigration(int partitionId)	Cancel an in-flight migration.
void	cleanupCommittedMigration(int partitionId)	Clean up source data for a migration that has already been staged and whose ownership cutover has been installed by the migration manager.
PartitionMigrationPipeline.MigrationTask	enqueueMigration(int partitionId, String sourceNode, String targetNode, long dataSize)	Enqueue a migration task to be processed by the pipeline.
Set<Integer>	getActiveMigrations()	Get all active migrations.
int	getPendingCount()	Get count of pending tasks in queue.
@Nullable PartitionMigrationPipeline.MigrationProgress	getProgress(int partitionId)	Get progress for a specific migration.
PartitionMigrationPipeline.MigrationStats	getStats()	Get current statistics.
@Nullable PartitionMigrationPipeline.MigrationTask	getTask(int partitionId)	Get the state of a specific migration task.
boolean	isPartitionMigrating(int partitionId)	Check if a partition is currently being migrated.
int	maxConcurrentMigrations()
void	processMigrations()	Execute migrations from the queue up to the concurrency limit.
void	processMigrationsSync()	Execute migrations synchronously (for testing).
void	rollbackStagedMigration(int partitionId)	Roll back a staged migration whose batch cutover failed before ownership was safely installed.

Methods inherited from class Object

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

Field Details

DEFAULT_CONCURRENT_LIMIT

public static final int DEFAULT_CONCURRENT_LIMIT

See Also:

• Constant Field Values

DEFAULT_MAX_RETRIES

public static final int DEFAULT_MAX_RETRIES

See Also:

• Constant Field Values

Constructor Details

PartitionMigrationPipeline

public PartitionMigrationPipeline(int instanceNumber,
 DataStructureRegistry registry,
 PartitionTable partitionTable)

PartitionMigrationPipeline

public PartitionMigrationPipeline(int instanceNumber,
 DataStructureRegistry registry,
 PartitionTable partitionTable,
 int maxConcurrent,
 int maxRetries)

PartitionMigrationPipeline

public PartitionMigrationPipeline(int instanceNumber,
 DataStructureRegistry registry,
 PartitionTable partitionTable,
 @Nullable PartitionMigrationPipeline.MigrationDataSender dataSender,
 @Nullable PartitionMigrationPipeline.OwnershipCommitter ownershipCommitter,
 int maxConcurrent,
 int maxRetries)

PartitionMigrationPipeline

public PartitionMigrationPipeline(int instanceNumber,
 DataStructureRegistry defaultRegistry,
 IntFunction<DataStructureRegistry> registryForSlot,
 PartitionTable partitionTable,
 @Nullable PartitionMigrationPipeline.MigrationDataSender dataSender,
 @Nullable PartitionMigrationPipeline.OwnershipCommitter ownershipCommitter,
 int maxConcurrent,
 int maxRetries)

Session 10 ESC-12.3: sharded-aware constructor for the historical one-registry-per-slot mode. Fully sharded callers should use the overload that supplies both the source registry collection and per-key target router.

PartitionMigrationPipeline

public PartitionMigrationPipeline(int instanceNumber,
 DataStructureRegistry defaultRegistry,
 IntFunction<DataStructureRegistry> registryForSlot,
 IntFunction<Collection<DataStructureRegistry>> registriesForSlot,
 Function<String, DataStructureRegistry> registryForKey,
 PartitionTable partitionTable,
 @Nullable PartitionMigrationPipeline.MigrationDataSender dataSender,
 @Nullable PartitionMigrationPipeline.OwnershipCommitter ownershipCommitter,
 int maxConcurrent,
 int maxRetries)

Method Details

maxConcurrentMigrations

public int maxConcurrentMigrations()

enqueueMigration

public PartitionMigrationPipeline.MigrationTask enqueueMigration(int partitionId,
 String sourceNode,
 String targetNode,
 long dataSize)

Enqueue a migration task to be processed by the pipeline.

Parameters:

partitionId - partition identifier (slot ID)

sourceNode - source node ID

targetNode - target node ID

dataSize - estimated size in bytes

Returns:

the created migration task

processMigrations

public void processMigrations()

Execute migrations from the queue up to the concurrency limit. Should be called regularly or triggered by a scheduler.

processMigrationsSync

public void processMigrationsSync()

Execute migrations synchronously (for testing). Processes all queued tasks in the calling thread.

cleanupCommittedMigration

public void cleanupCommittedMigration(int partitionId)

Clean up source data for a migration that has already been staged and whose ownership cutover has been installed by the migration manager.

rollbackStagedMigration

public void rollbackStagedMigration(int partitionId)

Roll back a staged migration whose batch cutover failed before ownership was safely installed. Source data is preserved; copied target data is removed if it still matches the staged snapshot.

isPartitionMigrating

public boolean isPartitionMigrating(int partitionId)

Check if a partition is currently being migrated.

Parameters:

partitionId - partition to check

Returns:

true if the partition is locked for migration

cancelMigration

public boolean cancelMigration(int partitionId)

Cancel an in-flight migration.

Parameters:

partitionId - partition to cancel

Returns:

true if cancellation was successful

getStats

public PartitionMigrationPipeline.MigrationStats getStats()

Get current statistics.

Returns:

migration statistics

getProgress

public @Nullable PartitionMigrationPipeline.MigrationProgress getProgress(int partitionId)

Get progress for a specific migration.

Parameters:

partitionId - partition identifier

Returns:

migration progress, or null if not found

getActiveMigrations

public Set<Integer> getActiveMigrations()

Get all active migrations.

Returns:

set of active partition IDs

getPendingCount

public int getPendingCount()

Get count of pending tasks in queue.

Returns:

number of pending migrations

getTask

public @Nullable PartitionMigrationPipeline.MigrationTask getTask(int partitionId)

Get the state of a specific migration task.

Parameters:

partitionId - partition identifier

Returns:

the current task, or null if no active migration for this partition