fix: block out of order migrations in migrations tool (#7693)

* fix: block out of order migrations in migrations tool * compute minimumVersion in loadMigrations function
confluentinc · Jun 23, 2021 · 1d617d3 · 1d617d3
1 parent 1823c0d
commit 1d617d3
Show file tree

Hide file tree

Showing 3 changed files with 33 additions and 5 deletions.
diff --git a/...ools/src/main/java/io/confluent/ksql/tools/migrations/commands/ApplyMigrationCommand.java b/...ools/src/main/java/io/confluent/ksql/tools/migrations/commands/ApplyMigrationCommand.java
@@ -201,14 +201,11 @@ private boolean apply(
       final Clock clock
   ) {
     String previous = MetadataUtil.getLatestMigratedVersion(config, ksqlClient);
-    final int minimumVersion = previous.equals(MetadataUtil.NONE_VERSION)
-        ? 1
-        : Integer.parseInt(previous) + 1;
 
     LOGGER.info("Loading migration files");
     final List<MigrationFile> migrations;
     try {
-      migrations = loadMigrationsToApply(migrationsDir, minimumVersion);
+      migrations = loadMigrationsToApply(migrationsDir, previous);
     } catch (MigrationException e) {
       LOGGER.error(e.getMessage());
       return false;
@@ -232,14 +229,22 @@ private boolean apply(
 
   private List<MigrationFile> loadMigrationsToApply(
       final String migrationsDir,
-      final int minimumVersion
+      final String previousVersion
   ) {
+    final int minimumVersion = previousVersion.equals(MetadataUtil.NONE_VERSION)
+        ? 1
+        : Integer.parseInt(previousVersion) + 1;
     if (version > 0) {
       final Optional<MigrationFile> migration =
           getMigrationForVersion(String.valueOf(version), migrationsDir);
       if (!migration.isPresent()) {
         throw new MigrationException("No migration file with version " + version + " exists.");
       }
+      if (version < minimumVersion) {
+        throw new MigrationException(
+            "Version must be newer than the last version migrated. Last version migrated was "
+                + previousVersion);
+      }
       return Collections.singletonList(migration.get());
     }
 

diff --git a/ksqldb-tools/src/test/java/io/confluent/ksql/tools/migrations/MigrationsTest.java b/ksqldb-tools/src/test/java/io/confluent/ksql/tools/migrations/MigrationsTest.java
@@ -266,6 +266,9 @@ private void shouldApplyMigrations() throws Exception {
     assertThat(applyStatus, is(0));
 
     verifyMigrationsApplied();
+
+    // Verify that older versions cannot be applied
+    assertThat(MIGRATIONS_CLI.parse("--config-file", configFilePath, "apply", "-v", "1").runCommand(), is(1));
   }
 
   private void shouldDisplayInfo() {

diff --git a/.../src/test/java/io/confluent/ksql/tools/migrations/commands/ApplyMigrationCommandTest.java b/.../src/test/java/io/confluent/ksql/tools/migrations/commands/ApplyMigrationCommandTest.java
@@ -641,6 +641,22 @@ public void shouldApplyDropConnectorStatement() throws Exception {
     inOrder.verifyNoMoreInteractions();
   }
 
+  @Test
+  public void shouldNotApplyOlderVersion() throws Exception {
+    // Given:
+    command = PARSER.parse("-v", "1");
+    createMigrationFile(1, NAME, migrationsDir, COMMAND);
+    givenAppliedMigration(1, NAME, MigrationState.MIGRATED);
+    givenCurrentMigrationVersion("1");
+
+    // When:
+    final int result = command.command(config, cfg -> ksqlClient, migrationsDir, Clock.fixed(
+        Instant.ofEpochMilli(1000), ZoneId.systemDefault()));
+
+    // Then:
+    assertThat(result, is(1));
+  }
+
   private void createMigrationFile(
       final int version,
       final String name,
@@ -718,6 +734,10 @@ private void givenAppliedMigration(
     when(row.getString(8)).thenReturn("no_error");
 
     when(infoQueryResult.get()).thenReturn(ImmutableList.of(row));
+    when(ksqlClient.executeQuery(
+        "SELECT version, checksum, previous, state, name, started_on, completed_on, error_reason FROM "
+            + MIGRATIONS_TABLE + " WHERE version_key = '" + version + "';"))
+        .thenReturn(infoQueryResult);
   }
 
   private void verifyMigratedVersion(