{"record":{"id":"8ea27b0e6d5a3f8b","repo":"apache/pulsar","slug":"migration-is-already-in-progress-phase-phase-8ea27b","errorCode":null,"errorMessage":"Migration is already in progress (phase: ${phase})","messagePattern":"Migration is already in progress \\(phase: (.+?)\\)","errorType":"exception","errorClass":"MetadataStoreException","httpStatus":null,"severity":"error","filePath":"pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/MigrationCoordinator.java","lineNumber":129,"sourceCode":"            throw e;\n        }\n    }\n\n    private void setInitialMigrationPhase() throws MetadataStoreException {\n        try {\n            Optional<GetResult> existing = sourceStore.get(MigrationState.MIGRATION_FLAG_PATH).get();\n            Optional<Long> expectedVersion;\n            if (existing.isEmpty()) {\n                // Create-only, to guard against concurrent migration starts\n                expectedVersion = Optional.of(-1L);\n            } else {\n                MigrationState currentState = ObjectMapperFactory.getMapper().reader()\n                        .readValue(existing.get().getValue(), MigrationState.class);\n                expectedVersion = switch (currentState.getPhase()) {\n                    // A leftover flag from a failed (or never started) migration can be replaced. The\n                    // expected version guards against concurrent migration starts.\n                    case NOT_STARTED, FAILED -> Optional.of(existing.get().getStat().getVersion());\n                    case PREPARATION, COPYING -> throw new MetadataStoreException(\n                            \"Migration is already in progress (phase: \" + currentState.getPhase() + \")\");\n                    case COMPLETED -> throw new MetadataStoreException(\"Migration has already been completed\");\n                };\n            }\n\n            sourceStore.put(MigrationState.MIGRATION_FLAG_PATH,\n                    ObjectMapperFactory.getMapper().writer()\n                            .writeValueAsBytes(new MigrationState(MigrationPhase.PREPARATION, targetUrl)),\n                    expectedVersion).get();\n        } catch (MetadataStoreException e) {\n            throw e;\n        } catch (Exception e) {\n            throw new MetadataStoreException(e);\n        }\n    }\n\n    private void updatePhase(MigrationPhase phase) throws MetadataStoreException {\n        try {","sourceCodeStart":111,"sourceCodeEnd":147,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/MigrationCoordinator.java#L111-L147","documentation":"startMigration() first writes a MigrationState flag at MIGRATION_FLAG_PATH on the source store. If a flag already exists with phase PREPARATION or COPYING, another migration is mid-flight, so setInitialMigrationPhase throws MetadataStoreException instead of clobbering the in-progress state. The expected-version guard exists precisely to prevent concurrent starts.","triggerScenarios":"Calling startMigration while a previous startMigration is in PREPARATION or COPYING phase (still running, hung, or its process died without advancing/resetting the flag).","commonSituations":"Re-running a migration script after a crashed attempt while the flag was left in PREPARATION; two operators starting the migration simultaneously.","solutions":["Check the MigrationState at MIGRATION_FLAG_PATH on the source store to see who/when set it","If the migration is genuinely stuck, manually reset the flag (delete or rewrite MIGRATION_FLAG_PATH with phase NOT_STARTED) and restart","Wait for the in-progress migration to reach COMPLETED or FAILED before starting a new one","Ensure only one operator/tool instance drives the migration at a time"],"exampleFix":"// before\ncoordinator.startMigration(); // throws: phase COPYING left over from crashed run\n// after\ncoordinator.resetMigrationFlag(); // set MigrationState(NOT_STARTED)\ncoordinator.startMigration();","handlingStrategy":"try-catch","validationCode":"// Java\nOptional<Versioned<byte[]>> flag =\n    sourceStore.get(MigrationState.MIGRATION_FLAG_PATH).get(30, TimeUnit.SECONDS);\nif (flag.isPresent()) {\n    MigrationState s = ObjectMapperFactory.getMapper().reader()\n        .readValue(flag.get().getValue(), MigrationState.class);\n    if (s.getPhase() == MigrationPhase.PREPARATION || s.getPhase() == MigrationPhase.COPYING) {\n        throw new IllegalStateException(\"Migration already running in phase \" + s.getPhase());\n    }\n}\n","typeGuard":"boolean migrationInProgress(MetadataStore src) throws Exception {\n    Optional<Versioned<byte[]>> f = src.get(MigrationState.MIGRATION_FLAG_PATH).get(30, TimeUnit.SECONDS);\n    return f.isPresent() && (readPhase(f.get()) == MigrationPhase.PREPARATION\n        || readPhase(f.get()) == MigrationPhase.COPYING);\n}\n","tryCatchPattern":"try {\n    coordinator.startMigration();\n} catch (MetadataStoreException e) {\n    if (e.getMessage().contains(\"already in progress\")) {\n        // inspect MIGRATION_FLAG_PATH; reset flag only if the run is confirmed dead\n    }\n}\n","preventionTips":["Serialize migration starts through a single operator/runbook","Always check the migration flag phase before invoking startMigration","After a crashed migration run, explicitly reset MIGRATION_FLAG_PATH to NOT_STARTED","Use the expected-version guard semantics — never force-write the flag while a migration may be live"],"tags":["migration","concurrency","metadata-store","oxia"],"backgroundTag":"migration-already-in-progress","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}