root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-16308 [4/4]: Add release-version flag to upgrade and downgrade commands (#17362) 

I've added the release-version flag to the upgrade and downgrade commands. I've also added tests.

While working on this, I realized that we reveal non-production features to be returned in the version-mapping and dependencies commands. I have changed this to only return production features (except in tests) and added tests for this.

Reviewers: Jun Rao <jun@confluent.io>
This commit is contained in:
Justine Olshan 2024-10-04 13:03:54 -07:00 committed by GitHub
parent c11a38f9df
commit c3f13b5c57
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 252 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
core/src/main/scala/kafka/tools/StorageTool.scala
View File

@ -88,11 +88,11 @@ object StorageTool extends Logging {
0 0
case "version-mapping" => case "version-mapping" =>
runVersionMappingCommand(namespace, printStream) runVersionMappingCommand(namespace, printStream, Features.PRODUCTION_FEATURES)
0 0
case "feature-dependencies" => case "feature-dependencies" =>
runFeatureDependenciesCommand(namespace, printStream) runFeatureDependenciesCommand(namespace, printStream, Features.PRODUCTION_FEATURES)
0 0
case "random-uuid" => case "random-uuid" =>
@ -151,10 +151,12 @@ object StorageTool extends Logging {
* *
* @param namespace Arguments containing the release version. * @param namespace Arguments containing the release version.
* @param printStream The print stream to output the version mapping. * @param printStream The print stream to output the version mapping.
* @param validFeatures List of features to be considered in the output
*/ */
def runVersionMappingCommand( def runVersionMappingCommand(
namespace: Namespace, namespace: Namespace,
printStream: PrintStream printStream: PrintStream,
validFeatures: java.util.List[Features]
): Unit = { ): Unit = {
val releaseVersion = Option(namespace.getString("release_version")).getOrElse(MetadataVersion.LATEST_PRODUCTION.toString) val releaseVersion = Option(namespace.getString("release_version")).getOrElse(MetadataVersion.LATEST_PRODUCTION.toString)
try { try {
@ -163,7 +165,7 @@ object StorageTool extends Logging {
val metadataVersionLevel = metadataVersion.featureLevel() val metadataVersionLevel = metadataVersion.featureLevel()
printStream.print(f"metadata.version=$metadataVersionLevel%d ($releaseVersion%s)%n") printStream.print(f"metadata.version=$metadataVersionLevel%d ($releaseVersion%s)%n")
for (feature <- Features.values()) { for (feature <- validFeatures.asScala) {
val featureLevel = feature.defaultValue(metadataVersion) val featureLevel = feature.defaultValue(metadataVersion)
printStream.print(f"${feature.featureName}%s=$featureLevel%d%n") printStream.print(f"${feature.featureName}%s=$featureLevel%d%n")
} }
@ -176,58 +178,57 @@ object StorageTool extends Logging {
def runFeatureDependenciesCommand( def runFeatureDependenciesCommand(
namespace: Namespace, namespace: Namespace,
printStream: PrintStream printStream: PrintStream,
validFeatures: java.util.List[Features]
): Unit = { ): Unit = {
val featureArgs = Option(namespace.getList[String]("feature")).map(_.asScala.toList).getOrElse(List.empty) val featureArgs = Option(namespace.getList[String]("feature")).map(_.asScala.toList).getOrElse(List.empty)
// Iterate over each feature specified with --feature // Iterate over each feature specified with --feature
if (featureArgs != null) { for (featureArg <- featureArgs) {
for (featureArg <- featureArgs) { val Array(featureName, versionStr) = featureArg.split("=")
val Array(featureName, versionStr) = featureArg.split("=")
val featureLevel = try { val featureLevel = try {
versionStr.toShort versionStr.toShort
} catch {
case _: NumberFormatException =>
throw new TerseFailure(s"Invalid version format: $versionStr for feature $featureName")
}
if (featureName == MetadataVersion.FEATURE_NAME) {
val metadataVersion = try {
MetadataVersion.fromFeatureLevel(featureLevel)
} catch { } catch {
case _: NumberFormatException => case _: IllegalArgumentException =>
throw new TerseFailure(s"Invalid version format: $versionStr for feature $featureName") throw new TerseFailure(s"Unknown metadata.version $featureLevel")
} }
printStream.printf("%s=%d (%s) has no dependencies.%n", featureName, featureLevel, metadataVersion.version())
} else {
validFeatures.asScala.find(_.featureName == featureName) match {
case Some(feature) =>
val featureVersion = try {
feature.fromFeatureLevel(featureLevel, true)
} catch {
case _: IllegalArgumentException =>
throw new TerseFailure(s"Feature level $featureLevel is not supported for feature $featureName")
}
val dependencies = featureVersion.dependencies().asScala
if (featureName == MetadataVersion.FEATURE_NAME) { if (dependencies.isEmpty) {
val metadataVersion = try { printStream.printf("%s=%d has no dependencies.%n", featureName, featureLevel)
MetadataVersion.fromFeatureLevel(featureLevel) } else {
} catch { printStream.printf("%s=%d requires:%n", featureName, featureLevel)
case _: IllegalArgumentException => for ((depFeature, depLevel) <- dependencies) {
throw new TerseFailure(s"Unknown metadata.version $featureLevel") if (depFeature == MetadataVersion.FEATURE_NAME) {
} val metadataVersion = MetadataVersion.fromFeatureLevel(depLevel)
printStream.printf("%s=%d (%s) has no dependencies.%n", featureName, featureLevel, metadataVersion.version()) printStream.println(s" $depFeature=$depLevel (${metadataVersion.version()})")
} else { } else {
Features.values().find(_.featureName == featureName) match { printStream.println(s" $depFeature=$depLevel")
case Some(feature) =>
val featureVersion = try {
feature.fromFeatureLevel(featureLevel, true)
} catch {
case _: IllegalArgumentException =>
throw new TerseFailure(s"Feature level $featureLevel is not supported for feature $featureName")
}
val dependencies = featureVersion.dependencies().asScala
if (dependencies.isEmpty) {
printStream.printf("%s=%d has no dependencies.%n", featureName, featureLevel)
} else {
printStream.printf("%s=%d requires:%n", featureName, featureLevel)
for ((depFeature, depLevel) <- dependencies) {
if (depFeature == MetadataVersion.FEATURE_NAME) {
val metadataVersion = MetadataVersion.fromFeatureLevel(depLevel)
printStream.println(s" $depFeature=$depLevel (${metadataVersion.version()})")
} else {
printStream.println(s" $depFeature=$depLevel")
}
} }
} }
}
case None => case None =>
throw new TerseFailure(s"Unknown feature: $featureName") throw new TerseFailure(s"Unknown feature: $featureName")
}
} }
} }
} }

87
core/src/test/scala/unit/kafka/tools/StorageToolTest.scala
View File

@ -39,7 +39,7 @@ import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.ValueSource import org.junit.jupiter.params.provider.ValueSource
import scala.collection.mutable.ListBuffer import scala.collection.mutable.ListBuffer
import scala.jdk.CollectionConverters.IterableHasAsScala import scala.jdk.CollectionConverters._
@Timeout(value = 40) @Timeout(value = 40)
class StorageToolTest { class StorageToolTest {
@ -54,7 +54,7 @@ class StorageToolTest {
properties properties
} }
val allFeatures = Features.FEATURES.toList val testingFeatures = Features.FEATURES.toList.asJava
@Test @Test
def testConfigToLogDirectories(): Unit = { def testConfigToLogDirectories(): Unit = {
@ -441,23 +441,17 @@ Found problem:
stream: ByteArrayOutputStream, stream: ByteArrayOutputStream,
releaseVersion: String releaseVersion: String
): Int = { ): Int = {
val tempDir = TestUtils.tempDir() // Prepare the arguments list
try { val arguments = ListBuffer[String]("version-mapping")
// Prepare the arguments list
val arguments = ListBuffer[String]("version-mapping")
// Add the release version argument // Add the release version argument
if (releaseVersion != null) { if (releaseVersion != null) {
arguments += "--release-version" arguments += "--release-version"
arguments += releaseVersion arguments += releaseVersion
}
// Execute the StorageTool with the arguments
StorageTool.execute(arguments.toArray, new PrintStream(stream))
} finally {
Utils.delete(tempDir)
} }
// Execute the StorageTool with the arguments
StorageTool.execute(arguments.toArray, new PrintStream(stream))
} }
@Test @Test
@ -473,7 +467,7 @@ Found problem:
s"Output did not contain expected Metadata Version: $output" s"Output did not contain expected Metadata Version: $output"
) )
for (feature <- Features.values()) { for (feature <- Features.PRODUCTION_FEATURES.asScala) {
val featureLevel = feature.defaultValue(metadataVersion) val featureLevel = feature.defaultValue(metadataVersion)
assertTrue(output.contains(s"${feature.featureName()}=$featureLevel"), assertTrue(output.contains(s"${feature.featureName()}=$featureLevel"),
s"Output did not contain expected feature mapping: $output" s"Output did not contain expected feature mapping: $output"
@ -496,7 +490,7 @@ Found problem:
s"Output did not contain expected Metadata Version: $output" s"Output did not contain expected Metadata Version: $output"
) )
for (feature <- Features.values()) { for (feature <- Features.PRODUCTION_FEATURES.asScala) {
val featureLevel = feature.defaultValue(metadataVersion) val featureLevel = feature.defaultValue(metadataVersion)
assertTrue(output.contains(s"${feature.featureName()}=$featureLevel"), assertTrue(output.contains(s"${feature.featureName()}=$featureLevel"),
s"Output did not contain expected feature mapping: $output" s"Output did not contain expected feature mapping: $output"
@ -534,37 +528,20 @@ Found problem:
stream: ByteArrayOutputStream, stream: ByteArrayOutputStream,
features: Seq[String] features: Seq[String]
): Int = { ): Int = {
val tempDir = TestUtils.tempDir() val arguments = ListBuffer[String]("feature-dependencies")
try { features.foreach(feature => {
val arguments = ListBuffer[String]("feature-dependencies") arguments += "--feature"
features.foreach(feature => { arguments += feature
arguments += "--feature" })
arguments += feature StorageTool.execute(arguments.toArray, new PrintStream(stream))
})
StorageTool.execute(arguments.toArray, new PrintStream(stream))
} finally {
Utils.delete(tempDir)
}
} }
@Test @Test
def testHandleFeatureDependenciesForFeatureWithDependencies(): Unit = { def testTestingFeatureDependencies(): Unit = {
val stream = new ByteArrayOutputStream() val stream = new ByteArrayOutputStream()
assertEquals(0, runFeatureDependenciesCommand(stream, Seq("test.feature.version=2"))) val namespace = StorageTool.parseArguments(Array("feature-dependencies", "--feature", "test.feature.version=2"))
val output = stream.toString StorageTool.runFeatureDependenciesCommand(namespace, new PrintStream(stream), testingFeatures)
val metadataVersion = MetadataVersion.latestTesting()
val expectedOutput = s"test.feature.version=2 requires:\n metadata.version=${metadataVersion.featureLevel()} (${metadataVersion.version()})\n"
assertEquals(expectedOutput.trim, output.trim)
}
@Test
def testMultipleFeatureDependencies(): Unit = {
val stream = new ByteArrayOutputStream()
val features = Seq("transaction.version=2", "group.version=1", "test.feature.version=2")
assertEquals(0, runFeatureDependenciesCommand(stream, features))
val output = stream.toString.trim val output = stream.toString.trim
System.out.println(output) System.out.println(output)
@ -572,11 +549,27 @@ Found problem:
val latestTestingVersion = MetadataVersion.latestTesting() val latestTestingVersion = MetadataVersion.latestTesting()
val latestTestingVersionString = s"metadata.version=${latestTestingVersion.featureLevel()} (${latestTestingVersion.version()})" val latestTestingVersionString = s"metadata.version=${latestTestingVersion.featureLevel()} (${latestTestingVersion.version()})"
val expectedOutput =
s"""test.feature.version=2 requires:
| $latestTestingVersionString
|""".stripMargin.trim
assertEquals(expectedOutput, output)
}
@Test
def testMultipleFeatureDependencies(): Unit = {
val stream = new ByteArrayOutputStream()
val features = Seq("transaction.version=2", "group.version=1")
assertEquals(0, runFeatureDependenciesCommand(stream, features))
val output = stream.toString.trim
System.out.println(output)
val expectedOutput = val expectedOutput =
s"""transaction.version=2 has no dependencies. s"""transaction.version=2 has no dependencies.
|group.version=1 has no dependencies. |group.version=1 has no dependencies.
|test.feature.version=2 requires:
| $latestTestingVersionString
|""".stripMargin.trim |""".stripMargin.trim
assertEquals(expectedOutput, output) assertEquals(expectedOutput, output)

89
tools/src/main/java/org/apache/kafka/tools/FeatureCommand.java
View File

@ -118,10 +118,10 @@ public class FeatureCommand {
handleDisable(namespace, adminClient); handleDisable(namespace, adminClient);
break; break;
case "version-mapping": case "version-mapping":
handleVersionMapping(namespace); handleVersionMapping(namespace, Features.PRODUCTION_FEATURES);
break; break;
case "feature-dependencies": case "feature-dependencies":
handleFeatureDependencies(namespace); handleFeatureDependencies(namespace, Features.PRODUCTION_FEATURES);
break; break;
default: default:
throw new TerseException("Unknown command " + command); throw new TerseException("Unknown command " + command);
@ -138,7 +138,11 @@ public class FeatureCommand {
Subparser upgradeParser = subparsers.addParser("upgrade") Subparser upgradeParser = subparsers.addParser("upgrade")
.help("Upgrade one or more feature flags."); .help("Upgrade one or more feature flags.");
upgradeParser.addArgument("--metadata") upgradeParser.addArgument("--metadata")
.help("The level to which we should upgrade the metadata. For example, 3.3-IV3.") .help("DEPRECATED -- The level to which we should upgrade the metadata. For example, 3.3-IV3.")
.action(store());
upgradeParser.addArgument("--release-version")
.help("The release version to update all features to. For example, 3.9-IV0 will set metadata.version=21 and kraft.version=1." +
" Use the version-mapping command to learn which features will be set for any given version.")
.action(store()); .action(store());
upgradeParser.addArgument("--feature") upgradeParser.addArgument("--feature")
.help("A feature upgrade we should perform, in feature=level format. For example: `metadata.version=5`.") .help("A feature upgrade we should perform, in feature=level format. For example: `metadata.version=5`.")
@ -153,7 +157,11 @@ public class FeatureCommand {
Subparser downgradeParser = subparsers.addParser("downgrade") Subparser downgradeParser = subparsers.addParser("downgrade")
.help("Upgrade one or more feature flags."); .help("Upgrade one or more feature flags.");
downgradeParser.addArgument("--metadata") downgradeParser.addArgument("--metadata")
.help("The level to which we should downgrade the metadata. For example, 3.3-IV0.") .help("DEPRECATED -- The level to which we should downgrade the metadata. For example, 3.3-IV0.")
.action(store());
downgradeParser.addArgument("--release-version")
.help("The release version to downgrade all features to. For example, 3.9-IV0 will set metadata.version=21 and kraft.version=1." +
" Use the version-mapping command to learn which features will be set for any given version.")
.action(store()); .action(store());
downgradeParser.addArgument("--feature") downgradeParser.addArgument("--feature")
.help("A feature downgrade we should perform, in feature=level format. For example: `metadata.version=5`.") .help("A feature downgrade we should perform, in feature=level format. For example: `metadata.version=5`.")
@ -272,30 +280,61 @@ public class FeatureCommand {
} }
private static void handleUpgradeOrDowngrade(String op, Namespace namespace, Admin admin, FeatureUpdate.UpgradeType upgradeType) throws TerseException { private static void handleUpgradeOrDowngrade(String op, Namespace namespace, Admin admin, FeatureUpdate.UpgradeType upgradeType) throws TerseException {
Map<String, FeatureUpdate> updates = new HashMap<>();
MetadataVersion version;
String metadata = namespace.getString("metadata"); String metadata = namespace.getString("metadata");
if (metadata != null) { List<String> features = namespace.getList("feature");
String releaseVersion = namespace.getString("release_version");
if (releaseVersion != null && (metadata != null || features != null)) {
throw new TerseException("Can not specify `release-version` with other feature flags.");
}
Map<String, FeatureUpdate> updates = new HashMap<>();
MetadataVersion metadataVersion;
if (releaseVersion != null) {
try { try {
version = MetadataVersion.fromVersionString(metadata); metadataVersion = MetadataVersion.fromVersionString(releaseVersion);
updates.put(metadataVersion.featureName(), new FeatureUpdate(metadataVersion.featureLevel(), upgradeType));
} catch (Throwable e) { } catch (Throwable e) {
throw new TerseException("Unknown metadata.version " + metadata + throw new TerseException("Unknown metadata.version " + releaseVersion +
". Supported metadata.version are " + metadataVersionsToString( ". Supported metadata.version are " + metadataVersionsToString(
MetadataVersion.MINIMUM_BOOTSTRAP_VERSION, MetadataVersion.latestProduction())); MetadataVersion.MINIMUM_BOOTSTRAP_VERSION, MetadataVersion.latestProduction()));
} }
updates.put(MetadataVersion.FEATURE_NAME, new FeatureUpdate(version.featureLevel(), upgradeType)); try {
} for (Features feature : Features.PRODUCTION_FEATURES) {
short featureLevel = feature.defaultValue(metadataVersion);
List<String> features = namespace.getList("feature"); // Don't send a request to upgrade a feature to 0.
if (features != null) { if (upgradeType != FeatureUpdate.UpgradeType.UPGRADE || featureLevel > 0) {
features.forEach(feature -> { updates.put(feature.featureName(), new FeatureUpdate(featureLevel, upgradeType));
String[] nameAndLevel; }
nameAndLevel = parseNameAndLevel(feature);
if (updates.put(nameAndLevel[0], new FeatureUpdate(Short.parseShort(nameAndLevel[1]), upgradeType)) != null) {
throw new RuntimeException("Feature " + nameAndLevel[0] + " was specified more than once.");
} }
}); } catch (Throwable e) {
throw new TerseException(upgradeType.name() + " for release version " + releaseVersion +
" failed because at least one feature had the following error: " + e.getMessage());
}
} else {
if (metadata != null) {
System.out.println(" `metadata` flag is deprecated and may be removed in a future release.");
try {
metadataVersion = MetadataVersion.fromVersionString(metadata);
} catch (Throwable e) {
throw new TerseException("Unknown metadata.version " + metadata +
". Supported metadata.version are " + metadataVersionsToString(
MetadataVersion.MINIMUM_BOOTSTRAP_VERSION, MetadataVersion.latestProduction()));
}
updates.put(MetadataVersion.FEATURE_NAME, new FeatureUpdate(metadataVersion.featureLevel(), upgradeType));
}
if (features != null) {
features.forEach(feature -> {
String[] nameAndLevel;
nameAndLevel = parseNameAndLevel(feature);
if (updates.put(nameAndLevel[0], new FeatureUpdate(Short.parseShort(nameAndLevel[1]), upgradeType)) != null) {
throw new RuntimeException("Feature " + nameAndLevel[0] + " was specified more than once.");
}
});
}
} }
update(op, admin, updates, namespace.getBoolean("dry_run")); update(op, admin, updates, namespace.getBoolean("dry_run"));
@ -317,7 +356,7 @@ public class FeatureCommand {
update("disable", adminClient, updates, namespace.getBoolean("dry_run")); update("disable", adminClient, updates, namespace.getBoolean("dry_run"));
} }
static void handleVersionMapping(Namespace namespace) throws TerseException { static void handleVersionMapping(Namespace namespace, List<Features> validFeatures) throws TerseException {
// Get the release version from the command-line arguments or default to the latest stable version // Get the release version from the command-line arguments or default to the latest stable version
String releaseVersion = Optional.ofNullable(namespace.getString("release_version")) String releaseVersion = Optional.ofNullable(namespace.getString("release_version"))
.orElseGet(() -> MetadataVersion.latestProduction().version()); .orElseGet(() -> MetadataVersion.latestProduction().version());
@ -328,7 +367,7 @@ public class FeatureCommand {
short metadataVersionLevel = version.featureLevel(); short metadataVersionLevel = version.featureLevel();
System.out.printf("metadata.version=%d (%s)%n", metadataVersionLevel, releaseVersion); System.out.printf("metadata.version=%d (%s)%n", metadataVersionLevel, releaseVersion);
for (Features feature : Features.values()) { for (Features feature : validFeatures) {
short featureLevel = feature.defaultValue(version); short featureLevel = feature.defaultValue(version);
System.out.printf("%s=%d%n", feature.featureName(), featureLevel); System.out.printf("%s=%d%n", feature.featureName(), featureLevel);
} }
@ -339,7 +378,7 @@ public class FeatureCommand {
} }
} }
static void handleFeatureDependencies(Namespace namespace) throws TerseException { static void handleFeatureDependencies(Namespace namespace, List<Features> validFeatures) throws TerseException {
List<String> featureArgs = namespace.getList("feature"); List<String> featureArgs = namespace.getList("feature");
// Iterate over each feature specified with --feature // Iterate over each feature specified with --feature
@ -361,7 +400,7 @@ public class FeatureCommand {
// Assuming metadata versions do not have dependencies. // Assuming metadata versions do not have dependencies.
System.out.printf("%s=%d (%s) has no dependencies.%n", featureName, featureLevel, metadataVersion.version()); System.out.printf("%s=%d (%s) has no dependencies.%n", featureName, featureLevel, metadataVersion.version());
} else { } else {
Features featureEnum = Arrays.stream(Features.FEATURES) Features featureEnum = validFeatures.stream()
.filter(f -> f.featureName().equals(featureName)) .filter(f -> f.featureName().equals(featureName))
.findFirst() .findFirst()
.orElseThrow(() -> new TerseException("Unknown feature: " + featureName)); .orElseThrow(() -> new TerseException("Unknown feature: " + featureName));

124
tools/src/test/java/org/apache/kafka/tools/FeatureCommandTest.java
View File

@ -49,6 +49,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
@ExtendWith(value = ClusterTestExtensions.class) @ExtendWith(value = ClusterTestExtensions.class)
public class FeatureCommandTest { public class FeatureCommandTest {
private final List<Features> testingFeatures = Arrays.stream(Features.FEATURES).collect(Collectors.toList());
@ClusterTest(types = {Type.KRAFT}, metadataVersion = MetadataVersion.IBP_3_3_IV1) @ClusterTest(types = {Type.KRAFT}, metadataVersion = MetadataVersion.IBP_3_3_IV1)
public void testDescribeWithKRaft(ClusterInstance cluster) { public void testDescribeWithKRaft(ClusterInstance cluster) {
String commandOutput = ToolsTestUtils.captureStandardOut(() -> String commandOutput = ToolsTestUtils.captureStandardOut(() ->
@ -100,7 +102,7 @@ public class FeatureCommandTest {
assertEquals(0, FeatureCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(), assertEquals(0, FeatureCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(),
"upgrade", "--metadata", "3.3-IV2")) "upgrade", "--metadata", "3.3-IV2"))
); );
assertEquals("metadata.version was upgraded to 6.", commandOutput); assertEquals(format("`metadata` flag is deprecated and may be removed in a future release.%nmetadata.version was upgraded to 6."), commandOutput);
} }
@ClusterTest(types = {Type.KRAFT}, metadataVersion = MetadataVersion.IBP_3_3_IV1) @ClusterTest(types = {Type.KRAFT}, metadataVersion = MetadataVersion.IBP_3_3_IV1)
@ -118,16 +120,62 @@ public class FeatureCommandTest {
"downgrade", "--metadata", "3.3-IV0")) "downgrade", "--metadata", "3.3-IV0"))
); );
assertEquals("Could not downgrade metadata.version to 4. The update failed for all features since the following " + assertEquals(format("`metadata` flag is deprecated and may be removed in a future release.%nCould not downgrade metadata.version to 4." +
"feature had an error: Invalid metadata.version 4. Refusing to perform the requested downgrade because it might delete metadata information.", commandOutput); " The update failed for all features since the following feature had an error: Invalid metadata.version 4." +
" Refusing to perform the requested downgrade because it might delete metadata information."), commandOutput);
commandOutput = ToolsTestUtils.captureStandardOut(() -> commandOutput = ToolsTestUtils.captureStandardOut(() ->
assertEquals(1, FeatureCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(), assertEquals(1, FeatureCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(),
"downgrade", "--unsafe", "--metadata", "3.3-IV0")) "downgrade", "--unsafe", "--metadata", "3.3-IV0"))
); );
assertEquals("Could not downgrade metadata.version to 4. The update failed for all features since the following " + assertEquals(format("`metadata` flag is deprecated and may be removed in a future release.%nCould not downgrade metadata.version to 4." +
"feature had an error: Invalid metadata.version 4. Unsafe metadata downgrade is not supported in this version.", commandOutput); " The update failed for all features since the following feature had an error: Invalid metadata.version 4." +
" Unsafe metadata downgrade is not supported in this version."), commandOutput);
}
@ClusterTest(types = {Type.KRAFT}, metadataVersion = MetadataVersion.IBP_3_8_IV0)
public void testUpgradeWithReleaseVersion(ClusterInstance cluster) {
String commandOutput = ToolsTestUtils.captureStandardOut(() ->
assertEquals(1, FeatureCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(),
"upgrade", "--release-version", "3.7-IV3"))
);
assertEquals("Could not upgrade metadata.version to 18. The update failed for all features since the following feature had an error:" +
" Invalid update version 18 for feature metadata.version. Can't downgrade the version of this feature without setting the upgrade type to either safe or unsafe downgrade.", commandOutput);
commandOutput = ToolsTestUtils.captureStandardOut(() ->
assertEquals(0, FeatureCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(),
"upgrade", "--release-version", "3.9-IV0"))
);
assertEquals("kraft.version was upgraded to 1.\n" +
"metadata.version was upgraded to 21.", commandOutput);
}
@ClusterTest(types = {Type.KRAFT}, metadataVersion = MetadataVersion.IBP_3_8_IV0)
public void testDowngradeWithReleaseVersion(ClusterInstance cluster) {
String commandOutput = ToolsTestUtils.captureStandardOut(() ->
assertEquals(1, FeatureCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(),
"downgrade", "--release-version", "3.9-IV0"))
);
assertTrue(commandOutput.contains("The update failed for all features since the following feature had an error:" +
" Invalid update version 1 for feature kraft.version. Can't downgrade to a newer version."));
assertTrue(commandOutput.contains("Could not downgrade group.version to 0."));
assertTrue(commandOutput.contains("Could not downgrade transaction.version to 0."));
assertTrue(commandOutput.contains("Could not downgrade kraft.version to 1."));
assertTrue(commandOutput.contains("Could not downgrade metadata.version to 21."));
commandOutput = ToolsTestUtils.captureStandardOut(() ->
assertEquals(0, FeatureCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(),
"downgrade", "--release-version", "3.7-IV3"))
);
assertEquals("group.version was downgraded to 0.\n" +
"kraft.version was downgraded to 0.\n" +
"metadata.version was downgraded to 18.\n" +
"transaction.version was downgraded to 0.", commandOutput);
} }
private String outputWithoutEpoch(String output) { private String outputWithoutEpoch(String output) {
@ -213,7 +261,8 @@ public class FeatureCommandTest {
Throwable t = assertThrows(TerseException.class, () -> FeatureCommand.handleUpgrade(new Namespace(namespace), buildAdminClient())); Throwable t = assertThrows(TerseException.class, () -> FeatureCommand.handleUpgrade(new Namespace(namespace), buildAdminClient()));
assertTrue(t.getMessage().contains("2 out of 2 operation(s) failed.")); assertTrue(t.getMessage().contains("2 out of 2 operation(s) failed."));
}); });
assertEquals(format("Could not upgrade foo.bar to 6. Invalid update version 5 for feature metadata.version. Can't upgrade to lower version.%n" + assertEquals(format("`metadata` flag is deprecated and may be removed in a future release.%nCould not upgrade foo.bar to 6." +
" Invalid update version 5 for feature metadata.version. Can't upgrade to lower version.%n" +
"Could not upgrade metadata.version to 5. Invalid update version 5 for feature metadata.version. Can't upgrade to lower version."), upgradeOutput); "Could not upgrade metadata.version to 5. Invalid update version 5 for feature metadata.version. Can't upgrade to lower version."), upgradeOutput);
} }
@ -227,7 +276,8 @@ public class FeatureCommandTest {
Throwable t = assertThrows(TerseException.class, () -> FeatureCommand.handleUpgrade(new Namespace(namespace), buildAdminClient())); Throwable t = assertThrows(TerseException.class, () -> FeatureCommand.handleUpgrade(new Namespace(namespace), buildAdminClient()));
assertTrue(t.getMessage().contains("2 out of 2 operation(s) failed.")); assertTrue(t.getMessage().contains("2 out of 2 operation(s) failed."));
}); });
assertEquals(format("Can not upgrade foo.bar to 6. Invalid update version 5 for feature metadata.version. Can't upgrade to lower version.%n" + assertEquals(format("`metadata` flag is deprecated and may be removed in a future release.%nCan not upgrade foo.bar to 6." +
" Invalid update version 5 for feature metadata.version. Can't upgrade to lower version.%n" +
"Can not upgrade metadata.version to 5. Invalid update version 5 for feature metadata.version. Can't upgrade to lower version."), upgradeOutput); "Can not upgrade metadata.version to 5. Invalid update version 5 for feature metadata.version. Can't upgrade to lower version."), upgradeOutput);
} }
@ -241,7 +291,8 @@ public class FeatureCommandTest {
Throwable t = assertThrows(TerseException.class, () -> FeatureCommand.handleDowngrade(new Namespace(namespace), buildAdminClient())); Throwable t = assertThrows(TerseException.class, () -> FeatureCommand.handleDowngrade(new Namespace(namespace), buildAdminClient()));
assertTrue(t.getMessage().contains("2 out of 2 operation(s) failed.")); assertTrue(t.getMessage().contains("2 out of 2 operation(s) failed."));
}); });
assertEquals(format("Could not downgrade foo.bar to 1. Invalid update version 7 for feature metadata.version. Can't downgrade to newer version.%n" + assertEquals(format("`metadata` flag is deprecated and may be removed in a future release.%nCould not downgrade foo.bar to 1." +
" Invalid update version 7 for feature metadata.version. Can't downgrade to newer version.%n" +
"Could not downgrade metadata.version to 7. Invalid update version 7 for feature metadata.version. Can't downgrade to newer version."), downgradeOutput); "Could not downgrade metadata.version to 7. Invalid update version 7 for feature metadata.version. Can't downgrade to newer version."), downgradeOutput);
} }
@ -255,8 +306,8 @@ public class FeatureCommandTest {
Throwable t = assertThrows(TerseException.class, () -> FeatureCommand.handleDowngrade(new Namespace(namespace), buildAdminClient())); Throwable t = assertThrows(TerseException.class, () -> FeatureCommand.handleDowngrade(new Namespace(namespace), buildAdminClient()));
assertTrue(t.getMessage().contains("2 out of 2 operation(s) failed.")); assertTrue(t.getMessage().contains("2 out of 2 operation(s) failed."));
}); });
assertEquals(format("Can not downgrade foo.bar to 1. Invalid update version 7 for feature metadata.version. Can't downgrade to newer version.%n" + assertEquals(format("`metadata` flag is deprecated and may be removed in a future release.%nCan not downgrade foo.bar to 1. Invalid update version 7 for feature metadata.version." +
"Can not downgrade metadata.version to 7. Invalid update version 7 for feature metadata.version. Can't downgrade to newer version."), downgradeOutput); " Can't downgrade to newer version.%nCan not downgrade metadata.version to 7. Invalid update version 7 for feature metadata.version. Can't downgrade to newer version."), downgradeOutput);
} }
@Test @Test
@ -287,13 +338,42 @@ public class FeatureCommandTest {
"Can not disable quux. Invalid update version 0 for feature metadata.version. Can't downgrade below 4"), disableOutput); "Can not disable quux. Invalid update version 0 for feature metadata.version. Can't downgrade below 4"), disableOutput);
} }
@Test
public void testInvalidReleaseVersion() {
Map<String, Object> namespace = new HashMap<>();
namespace.put("release_version", "foo");
ToolsTestUtils.captureStandardOut(() -> {
Throwable t = assertThrows(TerseException.class, () -> FeatureCommand.handleUpgrade(new Namespace(namespace), buildAdminClient()));
assertTrue(t.getMessage().contains("Unknown metadata.version foo."));
});
}
@Test
public void testIncompatibleUpgradeFlags() {
Map<String, Object> namespace = new HashMap<>();
namespace.put("release_version", "3.3-IV3");
namespace.put("feature", Arrays.asList("foo.bar", "metadata.version", "quux"));
ToolsTestUtils.captureStandardOut(() -> {
Throwable t = assertThrows(TerseException.class, () -> FeatureCommand.handleUpgrade(new Namespace(namespace), buildAdminClient()));
assertTrue(t.getMessage().contains("Can not specify `release-version` with other feature flags."));
});
namespace.put("release_version", "3.3-IV3");
namespace.put("metadata", "3.3-IV3");
ToolsTestUtils.captureStandardOut(() -> {
Throwable t = assertThrows(TerseException.class, () -> FeatureCommand.handleUpgrade(new Namespace(namespace), buildAdminClient()));
assertTrue(t.getMessage().contains("Can not specify `release-version` with other feature flags."));
});
}
@Test @Test
public void testHandleVersionMappingWithValidReleaseVersion() { public void testHandleVersionMappingWithValidReleaseVersion() {
Map<String, Object> namespace = new HashMap<>(); Map<String, Object> namespace = new HashMap<>();
namespace.put("release_version", "3.3-IV3"); namespace.put("release_version", "3.3-IV3");
String versionMappingOutput = ToolsTestUtils.captureStandardOut(() -> { String versionMappingOutput = ToolsTestUtils.captureStandardOut(() -> {
try { try {
FeatureCommand.handleVersionMapping(new Namespace(namespace)); FeatureCommand.handleVersionMapping(new Namespace(namespace), testingFeatures);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
@ -317,7 +397,7 @@ public class FeatureCommandTest {
Map<String, Object> namespace = new HashMap<>(); Map<String, Object> namespace = new HashMap<>();
String versionMappingOutput = ToolsTestUtils.captureStandardOut(() -> { String versionMappingOutput = ToolsTestUtils.captureStandardOut(() -> {
try { try {
FeatureCommand.handleVersionMapping(new Namespace(namespace)); FeatureCommand.handleVersionMapping(new Namespace(namespace), testingFeatures);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
@ -342,7 +422,7 @@ public class FeatureCommandTest {
namespace.put("release_version", "2.9-IV2"); namespace.put("release_version", "2.9-IV2");
TerseException exception1 = assertThrows(TerseException.class, () -> TerseException exception1 = assertThrows(TerseException.class, () ->
FeatureCommand.handleVersionMapping(new Namespace(namespace)) FeatureCommand.handleVersionMapping(new Namespace(namespace), testingFeatures)
); );
assertEquals("Unknown release version '2.9-IV2'." + assertEquals("Unknown release version '2.9-IV2'." +
@ -352,7 +432,7 @@ public class FeatureCommandTest {
namespace.put("release_version", "invalid"); namespace.put("release_version", "invalid");
TerseException exception2 = assertThrows(TerseException.class, () -> TerseException exception2 = assertThrows(TerseException.class, () ->
FeatureCommand.handleVersionMapping(new Namespace(namespace)) FeatureCommand.handleVersionMapping(new Namespace(namespace), testingFeatures)
); );
assertEquals("Unknown release version 'invalid'." + assertEquals("Unknown release version 'invalid'." +
@ -367,7 +447,7 @@ public class FeatureCommandTest {
String output = ToolsTestUtils.captureStandardOut(() -> { String output = ToolsTestUtils.captureStandardOut(() -> {
try { try {
FeatureCommand.handleFeatureDependencies(new Namespace(namespace)); FeatureCommand.handleFeatureDependencies(new Namespace(namespace), testingFeatures);
} catch (TerseException e) { } catch (TerseException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
@ -389,7 +469,7 @@ public class FeatureCommandTest {
String output = ToolsTestUtils.captureStandardOut(() -> { String output = ToolsTestUtils.captureStandardOut(() -> {
try { try {
FeatureCommand.handleFeatureDependencies(new Namespace(namespace)); FeatureCommand.handleFeatureDependencies(new Namespace(namespace), testingFeatures);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
@ -405,8 +485,8 @@ public class FeatureCommandTest {
Exception exception = assertThrows( Exception exception = assertThrows(
TerseException.class, TerseException.class,
() -> FeatureCommand.handleFeatureDependencies(new Namespace(namespace) () -> FeatureCommand.handleFeatureDependencies(new Namespace(namespace), testingFeatures)
)); );
assertEquals("Unknown feature: unknown.feature", exception.getMessage()); assertEquals("Unknown feature: unknown.feature", exception.getMessage());
} }
@ -418,8 +498,8 @@ public class FeatureCommandTest {
Exception exception = assertThrows( Exception exception = assertThrows(
IllegalArgumentException.class, IllegalArgumentException.class,
() -> FeatureCommand.handleFeatureDependencies(new Namespace(namespace) () -> FeatureCommand.handleFeatureDependencies(new Namespace(namespace), testingFeatures)
)); );
assertEquals("No feature:transaction.version with feature level 1000", exception.getMessage()); assertEquals("No feature:transaction.version with feature level 1000", exception.getMessage());
} }
@ -431,7 +511,7 @@ public class FeatureCommandTest {
RuntimeException exception = assertThrows( RuntimeException exception = assertThrows(
RuntimeException.class, RuntimeException.class,
() -> FeatureCommand.handleFeatureDependencies(new Namespace(namespace)) () -> FeatureCommand.handleFeatureDependencies(new Namespace(namespace), testingFeatures)
); );
assertEquals( assertEquals(
@ -451,7 +531,7 @@ public class FeatureCommandTest {
String output = ToolsTestUtils.captureStandardOut(() -> { String output = ToolsTestUtils.captureStandardOut(() -> {
try { try {
FeatureCommand.handleFeatureDependencies(new Namespace(namespace)); FeatureCommand.handleFeatureDependencies(new Namespace(namespace), testingFeatures);
} catch (TerseException e) { } catch (TerseException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }