fix(kafka): preserve stable log.dirs ordering across storageConfigs changes

pkg/resources/kafka/configmap.go

@@ -285,19 +285,36 @@ func configureBrokerZKMode(brokerID int32, kafkaCluster *v1beta1.KafkaCluster, c
 	}
 }
 
+// getEffectiveLogDirsMountPaths merges the previously configured log.dirs mount paths with the
+// currently desired ones, without reordering any path that already existed in mountPathsOld.
+// Kafka itself doesn't care about log.dirs ordering for partition placement, but in KRaft mode an
+// unset "metadata.log.dir" defaults to log.dirs[0], so gratuitously reshuffling the list (e.g. moving
+// a to-be-removed path to the end) could silently change which directory holds the metadata log.
+// To avoid that, existing paths keep their original relative order - including a path being kept
+// while its disk removal/rebalance is still pending - and only genuinely new paths (not present
+// before) are appended at the end.
 func getEffectiveLogDirsMountPaths(mountPathsOld, mountPathsNew []string, brokerID string, kafkaCluster *v1beta1.KafkaCluster) []string {
-	mountPathsEffective := append([]string{}, mountPathsNew...)
 	if len(mountPathsOld) == 0 {
-		return mountPathsEffective
+		return append([]string{}, mountPathsNew...)
 	}
 
 	newMountPathsSet := make(map[string]struct{}, len(mountPathsNew))
 	for _, path := range mountPathsNew {
 		newMountPathsSet[path] = struct{}{}
 	}
 
+	oldMountPathsSet := make(map[string]struct{}, len(mountPathsOld))
+	for _, path := range mountPathsOld {
+		oldMountPathsSet[path] = struct{}{}
+	}
+
+	mountPathsEffective := make([]string, 0, len(mountPathsOld)+len(mountPathsNew))
+
+	// Keep the existing order for paths already present in the old config, whether still
+	// declared or kept temporarily during a pending disk removal/rebalance.
 	for _, oldPath := range mountPathsOld {
-		if _, found := newMountPathsSet[oldPath]; found {
+		if _, stillDeclared := newMountPathsSet[oldPath]; stillDeclared {
+			mountPathsEffective = append(mountPathsEffective, oldPath)
 			continue
 		}
 
@@ -306,6 +323,14 @@ func getEffectiveLogDirsMountPaths(mountPathsOld, mountPathsNew []string, broker
 		}
 	}
 
+	// Append newly added paths (absent from the old config) at the end, in spec order.
+	for _, newPath := range mountPathsNew {
+		if _, alreadyPresent := oldMountPathsSet[newPath]; alreadyPresent {
+			continue
+		}
+		mountPathsEffective = append(mountPathsEffective, newPath)
+	}
+
 	return mountPathsEffective
 }
 

pkg/resources/kafka/configmap_test.go

@@ -216,6 +216,58 @@ func TestGetEffectiveLogDirsMountPaths(t *testing.T) {
 			},
 			expectedEffective: []string{"/kafka-logs/kafka", "/kafka-logs2/kafka"},
 		},
+		{
+			testName:          "reordering existing paths in storageConfigs does not reorder log.dirs",
+			mountPathsOld:     []string{"/kafka-logs/kafka", "/kafka-logs2/kafka", "/kafka-logs3/kafka"},
+			mountPathsNew:     []string{"/kafka-logs3/kafka", "/kafka-logs/kafka", "/kafka-logs2/kafka"},
+			brokerID:          "0",
+			kafkaCluster:      nil,
+			expectedEffective: []string{"/kafka-logs/kafka", "/kafka-logs2/kafka", "/kafka-logs3/kafka"},
+		},
+		{
+			testName:          "newly added paths are appended at the end in their declared relative order",
+			mountPathsOld:     []string{"/kafka-logs/kafka"},
+			mountPathsNew:     []string{"/kafka-logs2/kafka", "/kafka-logs/kafka", "/kafka-logs3/kafka"},
+			brokerID:          "0",
+			kafkaCluster:      nil,
+			expectedEffective: []string{"/kafka-logs/kafka", "/kafka-logs2/kafka", "/kafka-logs3/kafka"},
+		},
+		{
+			testName:          "path with pending disk removal re-declared in a swapped position keeps its original position",
+			mountPathsOld:     []string{"/kafka-logs/kafka", "/kafka-logs2/kafka"},
+			mountPathsNew:     []string{"/kafka-logs2/kafka", "/kafka-logs/kafka"},
+			brokerID:          "0",
+			kafkaCluster:      nil,
+			expectedEffective: []string{"/kafka-logs/kafka", "/kafka-logs2/kafka"},
+		},
+		{
+			testName:          "path re-added after its removal was already confirmed is appended at the end, ignoring its declared position",
+			mountPathsOld:     []string{"/kafka-logs2/kafka"},
+			mountPathsNew:     []string{"/kafka-logs/kafka", "/kafka-logs2/kafka"},
+			brokerID:          "0",
+			kafkaCluster:      nil,
+			expectedEffective: []string{"/kafka-logs2/kafka", "/kafka-logs/kafka"},
+		},
+		{
+			testName:      "mixed: pending-removal path keeps its position while a confirmed-removed-then-readded path is appended at the end",
+			mountPathsOld: []string{"/kafka-logs2/kafka", "/kafka-logs3/kafka"},
+			mountPathsNew: []string{"/kafka-logs/kafka", "/kafka-logs2/kafka"},
+			brokerID:      "0",
+			kafkaCluster: &v1beta1.KafkaCluster{
+				Status: v1beta1.KafkaClusterStatus{
+					BrokersState: map[string]v1beta1.BrokerState{
+						"0": {
+							GracefulActionState: v1beta1.GracefulActionState{
+								VolumeStates: map[string]v1beta1.VolumeState{
+									"/kafka-logs3": {CruiseControlVolumeState: v1beta1.GracefulDiskRemovalRequired},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedEffective: []string{"/kafka-logs2/kafka", "/kafka-logs3/kafka", "/kafka-logs/kafka"},
+		},
 	}
 	for _, test := range tests {
 		t.Run(test.testName, func(t *testing.T) {