Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Clean-up the KRaftUtils class #10987

Merged
merged 1 commit into from
Dec 26, 2024
Merged

Clean-up the KRaftUtils class #10987

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 0 additions & 56 deletions cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KRaftUtils.java
View file
Open in desktop

This file was deleted.

24 changes: 23 additions & 1 deletion cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaCluster.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@
import io.strimzi.operator.common.model.StatusUtils;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import org.apache.kafka.server.common.MetadataVersion;

import java.io.IOException;
import java.util.ArrayList;
Expand Down Expand Up @@ -305,7 +306,7 @@ public static KafkaCluster fromCrd(Reconciliation reconciliation,

// Validates and sets the metadata version used in KRaft
if (versionChange.metadataVersion() != null) {
KRaftUtils.validateMetadataVersion(versionChange.metadataVersion());
validateMetadataVersion(versionChange.metadataVersion());
result.metadataVersion = versionChange.metadataVersion();
}

Expand Down Expand Up @@ -544,6 +545,27 @@ public KafkaPool nodePoolForNodeId(int nodeId) {
throw new NodePoolNotFoundException("Node ID " + nodeId + " does not belong to any known node pool!");
}

/**
* Validates the metadata version
*
* @param metadataVersion Metadata version that should be validated
*/
/* test */ static void validateMetadataVersion(String metadataVersion) {
try {
MetadataVersion version = MetadataVersion.fromVersionString(metadataVersion);

// KRaft is supposed to be supported from metadata version 3.0-IV1. But only from metadata version 3.3-IV0,
// the initial metadata version can be set using the kafka-storage.sh utility. And since most metadata
// versions do not support downgrade, that means 3.3-IV0 is the oldest metadata version that can be used
// with Strimzi.
if (version.isLessThan(MetadataVersion.IBP_3_3_IV0)) {
throw new InvalidResourceException("The oldest supported metadata version is 3.3-IV0");
}
} catch (IllegalArgumentException e) {
throw new InvalidResourceException("Metadata version " + metadataVersion + " is invalid", e);
}
}

/**
* Validates the Kafka broker configuration against the configuration options of the desired Kafka version.
*
Expand Down
9 changes: 0 additions & 9 deletions ...or/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaAssemblyOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@
import io.strimzi.operator.cluster.ClusterOperatorConfig;
import io.strimzi.operator.cluster.PlatformFeaturesAvailability;
import io.strimzi.operator.cluster.model.ClusterCa;
import io.strimzi.operator.cluster.model.KRaftUtils;
import io.strimzi.operator.cluster.model.KafkaCluster;
import io.strimzi.operator.cluster.model.KafkaVersionChange;
import io.strimzi.operator.cluster.model.ModelUtils;
Expand All @@ -44,7 +43,6 @@
import io.strimzi.operator.common.Util;
import io.strimzi.operator.common.config.ConfigParameter;
import io.strimzi.operator.common.model.ClientsCa;
import io.strimzi.operator.common.model.InvalidResourceException;
import io.strimzi.operator.common.model.Labels;
import io.strimzi.operator.common.model.NamespaceAndName;
import io.strimzi.operator.common.model.PasswordGenerator;
Expand Down Expand Up @@ -232,13 +230,6 @@ Future<Void> reconcile(ReconciliationState reconcileState) {

if (nonMigratedCluster || !kraftEnabled || !nodePoolsEnabled) {
throw new InvalidConfigurationException("Strimzi " + OPERATOR_VERSION + " supports only KRaft-based Apache Kafka clusters. Please make sure your cluster is migrated to KRaft before using Strimzi " + OPERATOR_VERSION + ".");
} else {
// Validates features which are currently not supported in KRaft mode
try {
KRaftUtils.validateKafkaCrForKRaft(reconcileState.kafkaAssembly.getSpec());
} catch (InvalidResourceException e) {
return Future.failedFuture(e);
}
}

reconcileState.initialStatus()
Expand Down
71 changes: 0 additions & 71 deletions cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KRaftUtilsTest.java
View file
Open in desktop

This file was deleted.

26 changes: 25 additions & 1 deletion cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaClusterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2190,7 +2190,31 @@ public void testInvalidVersionWithCustomImage() {
}

@ParallelTest
public void testKRaftMetadataVersionValidation() {
public void testKRaftMetadataVersionValidation() {
// Valid values
assertDoesNotThrow(() -> KafkaCluster.validateMetadataVersion("3.6"));
assertDoesNotThrow(() -> KafkaCluster.validateMetadataVersion("3.6-IV2"));

// Minimum supported versions
assertDoesNotThrow(() -> KafkaCluster.validateMetadataVersion("3.3"));
assertDoesNotThrow(() -> KafkaCluster.validateMetadataVersion("3.3-IV0"));

// Invalid Values
InvalidResourceException e = assertThrows(InvalidResourceException.class, () -> KafkaCluster.validateMetadataVersion("3.6-IV9"));
assertThat(e.getMessage(), containsString("Metadata version 3.6-IV9 is invalid"));

e = assertThrows(InvalidResourceException.class, () -> KafkaCluster.validateMetadataVersion("3"));
assertThat(e.getMessage(), containsString("Metadata version 3 is invalid"));

e = assertThrows(InvalidResourceException.class, () -> KafkaCluster.validateMetadataVersion("3.2"));
assertThat(e.getMessage(), containsString("The oldest supported metadata version is 3.3-IV0"));

e = assertThrows(InvalidResourceException.class, () -> KafkaCluster.validateMetadataVersion("3.2-IV0"));
assertThat(e.getMessage(), containsString("The oldest supported metadata version is 3.3-IV0"));
}

@ParallelTest
public void testKRaftMetadataVersionValidationInFromCrd() {
Kafka kafka = new KafkaBuilder(KAFKA)
.editSpec()
.editKafka()
Expand Down
Loading