[kafka] branch 3.3 updated: KAFKA-14243: Temporarily disable unsafe downgrade (#12664)

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

davidarthur pushed a commit to branch 3.3
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/3.3 by this push:
     new ebf3cfe136 KAFKA-14243: Temporarily disable unsafe downgrade (#12664)
ebf3cfe136 is described below

commit ebf3cfe136bd75ade026da9ad22831dd4d6ce3d3
Author: Colin Patrick McCabe <cm...@apache.org>
AuthorDate: Tue Sep 20 12:32:52 2022 -0700

    KAFKA-14243: Temporarily disable unsafe downgrade (#12664)
    
    Reviewers: David Arthur <mu...@gmail.com>
---
 .../test/scala/unit/kafka/admin/FeatureCommandTest.scala  |  5 +++--
 .../apache/kafka/controller/FeatureControlManager.java    |  3 ++-
 .../kafka/controller/FeatureControlManagerTest.java       | 15 +++++++++++++++
 3 files changed, 20 insertions(+), 3 deletions(-)

diff --git a/core/src/test/scala/unit/kafka/admin/FeatureCommandTest.scala b/core/src/test/scala/unit/kafka/admin/FeatureCommandTest.scala
index 18884e4f93..3becc7b330 100644
--- a/core/src/test/scala/unit/kafka/admin/FeatureCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/FeatureCommandTest.scala
@@ -155,9 +155,10 @@ class FeatureCommandTest extends IntegrationTestHarness {
         "Retry using UNSAFE_DOWNGRADE if you want to force the downgrade to proceed.", env.outputWithoutEpoch())
     }
     TestUtils.resource(FeatureCommandTestEnv()) { env =>
-      assertEquals(0, FeatureCommand.mainNoExit(Array("--bootstrap-server", bootstrapServers(),
+      assertEquals(1, FeatureCommand.mainNoExit(Array("--bootstrap-server", bootstrapServers(),
         "downgrade", "--unsafe", "--metadata", "3.3-IV0"), env.out))
-      assertEquals("metadata.version was downgraded to 4.", env.outputWithoutEpoch())
+      assertEquals("Could not downgrade metadata.version to 4. Invalid metadata.version 4. " +
+        "Unsafe metadata downgrade is not supported in this version.", env.outputWithoutEpoch())
     }
   }
 }
diff --git a/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java
index 46f3be91b6..274e28906b 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java
@@ -255,7 +255,8 @@ public class FeatureControlManager {
             if (!metadataChanged) {
                 log.info("Downgrading metadata.version from {} to {}.", currentVersion, newVersion);
             } else if (allowUnsafeDowngrade) {
-                log.info("Downgrading metadata.version unsafely from {} to {}.", currentVersion, newVersion);
+                return invalidMetadataVersion(newVersionLevel, "Unsafe metadata downgrade is not supported " +
+                        "in this version.");
             } else {
                 return invalidMetadataVersion(newVersionLevel, "Refusing to perform the requested " +
                         "downgrade because it might delete metadata information. Retry using " +
diff --git a/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java
index 5162d879a5..3845f64666 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java
@@ -37,6 +37,7 @@ import org.apache.kafka.metadata.VersionRange;
 import org.apache.kafka.server.common.ApiMessageAndVersion;
 import org.apache.kafka.server.common.MetadataVersion;
 import org.apache.kafka.timeline.SnapshotRegistry;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 
@@ -326,6 +327,20 @@ public class FeatureControlManagerTest {
                 true));
     }
 
+    @Test
+    public void testUnsafeDowngradeIsTemporarilyDisabled() {
+        FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
+        assertEquals(ControllerResult.of(Collections.emptyList(),
+                        singletonMap(MetadataVersion.FEATURE_NAME, new ApiError(Errors.INVALID_UPDATE_VERSION,
+                                "Invalid metadata.version 4. Unsafe metadata downgrade is not supported in this version."))),
+                manager.updateFeatures(
+                        singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_3_IV0.featureLevel()),
+                        singletonMap(MetadataVersion.FEATURE_NAME, FeatureUpdate.UpgradeType.UNSAFE_DOWNGRADE),
+                        emptyMap(),
+                        true));
+    }
+
+    @Disabled
     @Test
     public void testCanUseUnsafeDowngradeIfMetadataChanged() {
         FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();