Add support for volumeAttributesClassName on persistent storage

api/src/main/java/io/strimzi/api/kafka/model/kafka/PersistentClaimStorage.java

@@ -26,7 +26,7 @@
         editableEnabled = false,
         builderPackage = Constants.FABRIC8_KUBERNETES_API
 )
-@JsonPropertyOrder({"id", "type", "size", "kraftMetadata", "class", "selector", "deleteClaim", "overrides"})
+@JsonPropertyOrder({"id", "type", "size", "kraftMetadata", "class", "volumeAttributesClass", "selector", "deleteClaim", "overrides"})
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 @EqualsAndHashCode(callSuper = true)
 @ToString(callSuper = true)
@@ -36,6 +36,7 @@ public class PersistentClaimStorage extends SingleVolumeStorage {
     private Map<String, String> selector;
     private boolean deleteClaim;
     private List<PersistentClaimStorageOverride> overrides;
+    private String volumeAttributesClass;
 
     @Description("Must be `" + TYPE_PERSISTENT_CLAIM + "`")
     @Override
@@ -125,4 +126,13 @@ public List<PersistentClaimStorageOverride> getOverrides() {
     public void setOverrides(List<PersistentClaimStorageOverride> overrides) {
         this.overrides = overrides;
     }
+
+    @Description("Specifies `VolumeAttributeClass` name for dynamically configuring storage attributes.")
+    public String getVolumeAttributesClass() {
+        return this.volumeAttributesClass;
+    }
+
+    public void setVolumeAttributesClass(String volumeAttributesClass) {
+        this.volumeAttributesClass = volumeAttributesClass;
+    }
 }

api/src/test/resources/crds/v1/045-Crd-kafkanodepool.yaml

@@ -98,6 +98,9 @@ spec:
                     - persistent-claim
                     - jbod
                     description: "Storage type, must be either 'ephemeral', 'persistent-claim', or 'jbod'."
+                  volumeAttributesClass:
+                    type: string
+                    description: Specifies `VolumeAttributeClass` name for dynamically configuring storage attributes.
                   volumes:
                     type: array
                     items:
@@ -136,6 +139,9 @@ spec:
                           - ephemeral
                           - persistent-claim
                           description: "Storage type, must be either 'ephemeral' or 'persistent-claim'."
+                        volumeAttributesClass:
+                          type: string
+                          description: Specifies `VolumeAttributeClass` name for dynamically configuring storage attributes.
                       required:
                       - type
                     description: List of volumes as Storage objects representing the JBOD disks array.

api/src/test/resources/crds/v1beta2/040-Crd-kafka.yaml

@@ -5396,6 +5396,8 @@ spec:
                         - ephemeral
                         - persistent-claim
                         - jbod
+                      volumeAttributesClass:
+                        type: string
                       volumes:
                         type: array
                         items:
@@ -5435,6 +5437,8 @@ spec:
                               enum:
                               - ephemeral
                               - persistent-claim
+                            volumeAttributesClass:
+                              type: string
                           required:
                           - type
                     required:
@@ -6944,6 +6948,8 @@ spec:
                         enum:
                         - ephemeral
                         - persistent-claim
+                      volumeAttributesClass:
+                        type: string
                     required:
                     - type
                   config:

api/src/test/resources/crds/v1beta2/045-Crd-kafkanodepool.yaml

@@ -89,6 +89,8 @@ spec:
                     - ephemeral
                     - persistent-claim
                     - jbod
+                  volumeAttributesClass:
+                    type: string
                   volumes:
                     type: array
                     items:
@@ -119,6 +121,8 @@ spec:
                           enum:
                           - ephemeral
                           - persistent-claim
+                        volumeAttributesClass:
+                          type: string
                       required:
                       - type
                 required:
@@ -1266,6 +1270,8 @@ spec:
                     - ephemeral
                     - persistent-claim
                     - jbod
+                  volumeAttributesClass:
+                    type: string
                   volumes:
                     type: array
                     items:
@@ -1305,6 +1311,8 @@ spec:
                           enum:
                           - ephemeral
                           - persistent-claim
+                        volumeAttributesClass:
+                          type: string
                       required:
                       - type
                 required:

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/PersistentVolumeClaimUtils.java

@@ -114,6 +114,7 @@ private static PersistentVolumeClaim createPersistentVolumeClaim(
                         .withRequests(requests)
                     .endResources()
                     .withStorageClassName(storage.getStorageClass())
+                    .withVolumeAttributesClassName(storage.getVolumeAttributesClass())
                     .withSelector(storageSelector)
                     .withVolumeMode("Filesystem")
                 .endSpec()

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/StorageDiff.java

@@ -28,7 +28,7 @@ public class StorageDiff extends AbstractJsonDiff {
     private static final ReconciliationLogger LOGGER = ReconciliationLogger.create(StorageDiff.class.getName());
 
     private static final Pattern IGNORABLE_PATHS = Pattern.compile(
-            "^(/deleteClaim|/kraftMetadata|/overrides.*|/)$");
+            "^(/deleteClaim|/kraftMetadata|/overrides.*|/volumeAttributesClass|/)$");
 
     private final boolean isEmpty;
     private final boolean changesType;

cluster-operator/src/test/java/io/strimzi/operator/cluster/model/PersistentVolumeClaimUtilsTest.java

@@ -45,6 +45,7 @@ public class PersistentVolumeClaimUtilsTest {
             .build();
     private final static PersistentClaimStorage PERSISTENT_CLAIM_STORAGE = new PersistentClaimStorageBuilder()
             .withStorageClass("my-storage-class")
+            .withVolumeAttributesClass("my-volume-attributes-class")
             .withSize("100Gi")
             .build();
     private final static Set<NodeRef> SINGLE_NODE = Set.of(new NodeRef(NAME + "-" + 0, 0, null, false, true));
@@ -95,6 +96,7 @@ public void testPersistentClaimStorage()  {
         assertThat(pvcs.get(0).getSpec().getSelector(), is(nullValue()));
         assertThat(pvcs.get(0).getSpec().getResources().getRequests(), is(Map.of("storage", new Quantity("100Gi", null))));
         assertThat(pvcs.get(0).getSpec().getStorageClassName(), is("my-storage-class"));
+        assertThat(pvcs.get(0).getSpec().getVolumeAttributesClassName(), is("my-volume-attributes-class"));
     }
 
     @Test

cluster-operator/src/test/java/io/strimzi/operator/cluster/model/StorageDiffTest.java

@@ -361,4 +361,42 @@ public void testDuplicateVolumeIds()    {
         assertThat(diff.isDuplicateVolumeIds(), is(true));
         assertThat(diff.issuesDetected(), is(true));
     }
+
+    @Test
+    public void testVolumeAttributesChanges() {
+        Storage persistent = new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withDeleteClaim(false).withId(0).withSize("100Gi").build();
+        Storage persistent2 = new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withDeleteClaim(false).withId(0).withSize("100Gi").withVolumeAttributesClass("vac-example").build();
+        Storage persistent3 = new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withDeleteClaim(false).withId(0).withSize("100Gi").withVolumeAttributesClass("vac-example-2").build();
+
+        StorageDiff diff = new StorageDiff(Reconciliation.DUMMY_RECONCILIATION, persistent, persistent, Set.of(0, 1, 2), Set.of(0, 1, 2));
+        assertThat(diff.issuesDetected(), is(false));
+
+        diff = new StorageDiff(Reconciliation.DUMMY_RECONCILIATION, persistent, persistent2, Set.of(0, 1, 2), Set.of(0, 1, 2));
+        assertThat(diff.issuesDetected(), is(false));
+
+        diff = new StorageDiff(Reconciliation.DUMMY_RECONCILIATION, persistent2, persistent3, Set.of(0, 1, 2), Set.of(0, 1, 2));
+        assertThat(diff.issuesDetected(), is(false));
+
+        Storage jbod = new JbodStorageBuilder().withVolumes(
+                new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withId(0).withVolumeAttributesClass("vac-example-1").build(),
+                new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withId(1).withVolumeAttributesClass("vac-example-2").build()
+        ).build();
+        Storage jbod2 = new JbodStorageBuilder().withVolumes(
+                new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withId(0).build(),
+                new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withId(1).withVolumeAttributesClass("vac-example-2").build()
+        ).build();
+        Storage jbod3 = new JbodStorageBuilder().withVolumes(
+                new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withId(0).withVolumeAttributesClass("vac-example-2").build(),
+                new PersistentClaimStorageBuilder().withStorageClass("gp2-ssd").withId(1).withVolumeAttributesClass("vac-example-2").build()
+        ).build();
+
+        diff = new StorageDiff(Reconciliation.DUMMY_RECONCILIATION, jbod, jbod, Set.of(0, 1, 2), Set.of(0, 1, 2));
+        assertThat(diff.issuesDetected(), is(false));
+
+        diff = new StorageDiff(Reconciliation.DUMMY_RECONCILIATION, jbod, jbod2, Set.of(0, 1, 2), Set.of(0, 1, 2));
+        assertThat(diff.issuesDetected(), is(false));
+
+        diff = new StorageDiff(Reconciliation.DUMMY_RECONCILIATION, jbod, jbod3, Set.of(0, 1, 2), Set.of(0, 1, 2));
+        assertThat(diff.issuesDetected(), is(false));
+    }
 }

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/JbodStorageMockTest.java

@@ -39,7 +39,6 @@
 import io.vertx.junit5.Checkpoint;
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
-import org.hamcrest.Matchers;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
@@ -54,10 +53,14 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
 
 @ExtendWith(VertxExtension.class)
 public class JbodStorageMockTest {
@@ -90,6 +93,7 @@ public static void beforeAll() {
                 .withPodController()
                 .withServiceController()
                 .withDeletionController()
+                .withPvcController()
                 .build();
         mockKube.start();
         client = mockKube.client();
@@ -190,7 +194,7 @@ public void testJbodStorageCreatesPVCsMatchingKafkaVolumes(VertxTestContext cont
                                     .filter(pvc -> pvc.getMetadata().getName().equals(expectedPvcName))
                                     .collect(Collectors.toList());
                             assertThat("Exactly one pvc should have the name " + expectedPvcName + " in :\n" + pvcs,
-                                    matchingPvcs, Matchers.hasSize(1));
+                                    matchingPvcs, hasSize(1));
 
                             PersistentVolumeClaim pvc = matchingPvcs.get(0);
                             boolean isDeleteClaim = ((PersistentClaimStorage) volume).isDeleteClaim();
@@ -320,6 +324,78 @@ public void testReconcileWithUpdateVolumeIdJbod(VertxTestContext context) {
             })));
     }
 
+    @Test
+    public void testReconcileWithUpdateVolumeAttributeClass(VertxTestContext context) {
+        assertThat(volumes.get(0), is(instanceOf(PersistentClaimStorage.class)));
+        PersistentClaimStorage storage = (PersistentClaimStorage) volumes.get(0);
+        storage.setVolumeAttributesClass("vac-example");
+        volumes.set(0, storage);
+
+        KafkaNodePool pool = new KafkaNodePoolBuilder(kafkaNodePool)
+            .editSpec()
+                .withNewJbodStorage()
+                    .withVolumes(volumes)
+                .endJbodStorage()
+            .endSpec()
+            .build();
+
+        Set<String> expectedPvcs = expectedPvcs(kafkaNodePool);
+
+        operator.reconcile(new Reconciliation("test-trigger", Kafka.RESOURCE_KIND, namespace, NAME))
+                .onComplete(context.succeeding(v -> context.verify(() -> {
+                    List<PersistentVolumeClaim> pvcs = getPvcs();
+                    Set<String> pvcsNames = pvcs.stream().map(pvc -> pvc.getMetadata().getName())
+                        .collect(Collectors.toSet());
+                    assertThat(pvcsNames, is(expectedPvcs));
+                })))
+                .compose(v -> {
+                    client.persistentVolumeClaims()
+                        .waitUntilCondition(pvc -> "Bound".equals(pvc.getStatus().getPhase()),
+                            5, TimeUnit.SECONDS);
+                    Crds.kafkaNodePoolOperation(client).inNamespace(namespace).withName(NODE_POOL_NAME).patch(pool);
+                    return operator.reconcile(new Reconciliation("vac-added-trigger", Kafka.RESOURCE_KIND, namespace, NAME));
+                })
+                .onComplete(context.succeeding(v -> context.verify(() -> {
+                    List<PersistentVolumeClaim> pvcs = getPvcs();
+                    Set<String> vacPvc = pvcs.stream()
+                        .filter(pvc -> "vac-example".equals(pvc.getSpec().getVolumeAttributesClassName()))
+                        .map(pvc -> pvc.getMetadata().getName())
+                        .collect(Collectors.toSet());
+                    assertThat(vacPvc, hasSize(3));
+                    assertThat(vacPvc, hasItem("data-0-my-cluster-mixed-0"));
+                    assertThat(vacPvc, hasItem("data-0-my-cluster-mixed-1"));
+                    assertThat(vacPvc, hasItem("data-0-my-cluster-mixed-2"));
+                })))
+                .compose(v -> {
+                    storage.setVolumeAttributesClass("vac-new");
+                    volumes.set(0, storage);
+                    KafkaNodePool pool1 = new KafkaNodePoolBuilder(pool)
+                        .editSpec()
+                            .withNewJbodStorage()
+                                .withVolumes(volumes)
+                            .endJbodStorage()
+                        .endSpec()
+                        .build();
+                    client.persistentVolumeClaims()
+                        .waitUntilCondition(pvc -> "Bound".equals(pvc.getStatus().getPhase()),
+                            5, TimeUnit.SECONDS);
+                    Crds.kafkaNodePoolOperation(client).inNamespace(namespace).withName(NODE_POOL_NAME).patch(pool1);
+                    return operator.reconcile(new Reconciliation("vac-updated-trigger", Kafka.RESOURCE_KIND, namespace, NAME));
+                })
+                .onComplete(context.succeeding(v -> context.verify(() -> {
+                    List<PersistentVolumeClaim> pvcs = getPvcs();
+                    Set<String> vacPvc = pvcs.stream()
+                        .filter(pvc -> "vac-new".equals(pvc.getSpec().getVolumeAttributesClassName()))
+                        .map(pvc -> pvc.getMetadata().getName())
+                        .collect(Collectors.toSet());
+                    assertThat(vacPvc, hasSize(3));
+                    assertThat(vacPvc, hasItem("data-0-my-cluster-mixed-0"));
+                    assertThat(vacPvc, hasItem("data-0-my-cluster-mixed-1"));
+                    assertThat(vacPvc, hasItem("data-0-my-cluster-mixed-2"));
+                    context.completeNow();
+                })));
+    }
+
     private Set<String> expectedPvcs(KafkaNodePool nodePool) {
         Set<String> expectedPvcs = new HashSet<>();
         for (int i = 0; i < nodePool.getSpec().getReplicas(); i++) {

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaAssemblyOperatorTest.java

@@ -245,7 +245,13 @@ public static Iterable<Params> data() {
                     .withSize("123")
                     .withStorageClass("foo")
                     .withDeleteClaim(true)
-                .build()
+                .build(),
+            new PersistentClaimStorageBuilder()
+                    .withSize("123")
+                    .withStorageClass("foo")
+                    .withVolumeAttributesClass("fooVac")
+                    .withDeleteClaim(true)
+                .build(),
         };
 
         List<Map<String, Object>> configs = asList(

documentation/modules/appendix_crds.adoc

@@ -681,6 +681,9 @@ It must have the value `persistent-claim` for the type `PersistentClaimStorage`.
 |class
 |string
 |The storage class to use for dynamic volume allocation.
+|volumeAttributesClass
+|string
+|Specifies `VolumeAttributeClass` name for dynamically configuring storage attributes.
 |selector
 |map
 |Specifies a specific persistent volume to use. It contains key:value pairs representing labels for selecting such a volume.

mockkube/src/main/java/io/strimzi/test/mockkube3/MockKube3.java

@@ -24,6 +24,7 @@
 import io.strimzi.test.mockkube3.controllers.MockDeletionController;
 import io.strimzi.test.mockkube3.controllers.MockDeploymentController;
 import io.strimzi.test.mockkube3.controllers.MockPodController;
+import io.strimzi.test.mockkube3.controllers.MockPvcController;
 import io.strimzi.test.mockkube3.controllers.MockServiceController;
 import org.testcontainers.utility.DockerImageName;
 
@@ -359,6 +360,16 @@ public MockKube3Builder withNamespaces(String... namespaces)  {
             return this;
         }
 
+        /**
+         * Registers PVC Controller to manage binding the PVC.
+         *
+         * @return  MockKube builder instance
+         */
+        public MockKube3Builder withPvcController() {
+            mock.registerController(new MockPvcController());
+            return this;
+        }
+
         /**
          * Builds an instance of MockKube based on the builder configuration
          *