@SuppressWarnings({"checkstyle:npathcomplexity", "checkstyle:cyclomaticcomplexity", "checkstyle:methodlength"})
    void planMigrations(int partitionId, PartitionReplica[] oldReplicas, PartitionReplica[] newReplicas,
            MigrationDecisionCallback callback) {
        assert oldReplicas.length == newReplicas.length : "Replica addresses with different lengths! Old: "
                + Arrays.toString(oldReplicas) + ", New: " + Arrays.toString(newReplicas);

        if (logger.isFinestEnabled()) {
            logger.finest("partitionId=%d, Initial state: %s", partitionId, Arrays.toString(oldReplicas));
            logger.finest("partitionId=%d, Final state: %s", partitionId, Arrays.toString(newReplicas));
        }

        initState(oldReplicas);
        assertNoDuplicate(partitionId, oldReplicas, newReplicas);

        // fix cyclic partition replica movements
        if (fixCycle(oldReplicas, newReplicas)) {
            if (logger.isFinestEnabled()) {
                logger.finest("partitionId=%d, Final state (after cycle fix): %s", partitionId,
                        Arrays.toString(newReplicas));
            }
        }

        int currentIndex = 0;
        while (currentIndex < oldReplicas.length) {
            if (logger.isFinestEnabled()) {
                logger.finest("partitionId=%d, Current index: %d, state: %s", partitionId, currentIndex,
                        Arrays.toString(state));
            }
            assertNoDuplicate(partitionId, oldReplicas, newReplicas);

            if (newReplicas[currentIndex] == null) {
                if (state[currentIndex] != null) {
                    // replica owner is removed and no one will own this replica
                    logger.finest("partitionId=%d, New address is null at index: %d", partitionId, currentIndex);
                    callback.migrate(state[currentIndex], currentIndex, -1, null, -1, -1);
                    state[currentIndex] = null;
                }
                currentIndex++;
                continue;
            }

            if (state[currentIndex] == null) {
                int i = getReplicaIndex(state, newReplicas[currentIndex]);
                if (i == -1) {
                    // fresh replica copy is needed, so COPY replica to newReplicas[currentIndex] from partition owner
                    logger.finest("partitionId=%d, COPY %s to index: %d", partitionId, newReplicas[currentIndex], currentIndex);
                    callback.migrate(null, -1, -1, newReplicas[currentIndex], -1, currentIndex);
                    state[currentIndex] = newReplicas[currentIndex];
                    currentIndex++;
                    continue;
                }

                if (i > currentIndex) {
                    // SHIFT UP replica from i to currentIndex, copy data from partition owner
                    logger.finest("partitionId=%d, SHIFT UP-2 %s from old addresses index: %d to index: %d", partitionId,
                            state[i], i, currentIndex);
                    callback.migrate(null, -1, -1, state[i], i, currentIndex);
                    state[currentIndex] = state[i];
                    state[i] = null;
                    continue;
                }

                throw new AssertionError("partitionId=" + partitionId
                        + "Migration decision algorithm failed during SHIFT UP! INITIAL: " + Arrays.toString(oldReplicas)
                        + ", CURRENT: " + Arrays.toString(state) + ", FINAL: " + Arrays.toString(newReplicas));
            }

            if (newReplicas[currentIndex].equals(state[currentIndex])) {
                // no change, no action needed
                currentIndex++;
                continue;
            }

            if (getReplicaIndex(newReplicas, state[currentIndex]) == -1
                    && getReplicaIndex(state, newReplicas[currentIndex]) == -1) {
                // MOVE partition replica from its old owner to new owner
                logger.finest("partitionId=%d, MOVE %s to index: %d", partitionId, newReplicas[currentIndex], currentIndex);
                callback.migrate(state[currentIndex], currentIndex, -1, newReplicas[currentIndex], -1, currentIndex);
                state[currentIndex] = newReplicas[currentIndex];
                currentIndex++;
                continue;
            }

            if (getReplicaIndex(state, newReplicas[currentIndex]) == -1) {
                int newIndex = getReplicaIndex(newReplicas, state[currentIndex]);
                assert newIndex > currentIndex : "partitionId=" + partitionId
                        + ", Migration decision algorithm failed during SHIFT DOWN! INITIAL: "
                        + Arrays.toString(oldReplicas) + ", CURRENT: " + Arrays.toString(state)
                        + ", FINAL: " + Arrays.toString(newReplicas);

                if (state[newIndex] == null) {
                    // it is a SHIFT DOWN
                    logger.finest("partitionId=%d, SHIFT DOWN %s to index: %d, COPY %s to index: %d", partitionId,
                            state[currentIndex], newIndex, newReplicas[currentIndex], currentIndex);
                    callback.migrate(state[currentIndex], currentIndex, newIndex, newReplicas[currentIndex], -1, currentIndex);
                    state[newIndex] = state[currentIndex];
                } else {
                    logger.finest("partitionId=%d, MOVE-3 %s to index: %d", partitionId, newReplicas[currentIndex], currentIndex);
                    callback.migrate(state[currentIndex], currentIndex, -1, newReplicas[currentIndex], -1, currentIndex);
                }

                state[currentIndex] = newReplicas[currentIndex];
                currentIndex++;
                continue;
            }

            planMigrations(partitionId, oldReplicas, newReplicas, callback, currentIndex);
        }

        assert Arrays.equals(state, newReplicas)
                : "partitionId=" + partitionId + ", Migration decisions failed! INITIAL: " + Arrays.toString(oldReplicas)
                + " CURRENT: " + Arrays.toString(state) + ", FINAL: " + Arrays.toString(newReplicas);
    }
@Test
    public void test_SHIFT_DOWN_withNullKeepReplicaIndex() throws UnknownHostException {
        final PartitionReplica[] oldReplicas = {
                new PartitionReplica(new Address("localhost", 5701), uuids[0]),
                null,
                new PartitionReplica(new Address("localhost", 5703), uuids[2]),
                null,
                null,
                null,
                null,
        };
        final PartitionReplica[] newReplicas = {
                new PartitionReplica(new Address("localhost", 5704), uuids[3]),
                new PartitionReplica(new Address("localhost", 5701), uuids[0]),
                new PartitionReplica(new Address("localhost", 5703), uuids[2]),
                null,
                null,
                null,
                null,
        };

        migrationPlanner.planMigrations(0, oldReplicas, newReplicas, callback);
        verify(callback).migrate(new PartitionReplica(new Address("localhost", 5701), uuids[0]), 0, 1, new PartitionReplica(new Address("localhost", 5704), uuids[3]), -1, 0);
    }