KAFKA-20921: Make ScramImage mechanism maps unmodifiable

metadata/src/main/java/org/apache/kafka/image/ScramImage.java

@@ -46,7 +46,11 @@ public record ScramImage(Map<ScramMechanism, Map<String, ScramCredentialData>> m
     public static final ScramImage EMPTY = new ScramImage(Map.of());
 
     public ScramImage {
-        mechanisms = Collections.unmodifiableMap(mechanisms);
+        Map<ScramMechanism, Map<String, ScramCredentialData>> wrapped = new HashMap<>(mechanisms.size());
+        for (var entry : mechanisms.entrySet()) {
+            wrapped.put(entry.getKey(), Collections.unmodifiableMap(entry.getValue()));
+        }
+        mechanisms = Collections.unmodifiableMap(wrapped);
     }
 
     public void write(ImageWriter writer, ImageWriterOptions options) {

metadata/src/main/java/org/apache/kafka/metadata/ScramCredentialData.java

@@ -25,7 +25,7 @@
 import java.util.Objects;
 
 /**
- * Represents the ACLs in the metadata image.
+ * Represents SCRAM credential data in the metadata image.
  * <p>
  * This class is thread-safe.
  */

metadata/src/test/java/org/apache/kafka/image/ScramImageTest.java

@@ -17,16 +17,20 @@
 
 package org.apache.kafka.image;
 
+import org.apache.kafka.clients.admin.ScramMechanism;
 import org.apache.kafka.image.writer.ImageWriterOptions;
 import org.apache.kafka.image.writer.RecordListWriter;
 import org.apache.kafka.metadata.RecordTestUtils;
+import org.apache.kafka.metadata.ScramCredentialData;
 import org.apache.kafka.server.common.ApiMessageAndVersion;
 import org.apache.kafka.server.common.MetadataVersion;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -67,6 +71,22 @@ public void testImage2RoundTrip() {
         testToImage(IMAGE2);
     }
 
+    @Test
+    public void testMechanismMapsAreUnmodifiable() {
+        ScramCredentialData credential = new ScramCredentialData(
+            new byte[] {1}, new byte[] {2}, new byte[] {3}, 4096);
+        Map<String, ScramCredentialData> credentials = new HashMap<>();
+        credentials.put("alice", credential);
+        Map<ScramMechanism, Map<String, ScramCredentialData>> mechanisms = new HashMap<>();
+        mechanisms.put(ScramMechanism.SCRAM_SHA_256, credentials);
+
+        ScramImage image = new ScramImage(mechanisms);
+
+        assertThrows(UnsupportedOperationException.class, () -> image.mechanisms().clear());
+        assertThrows(UnsupportedOperationException.class,
+            () -> image.mechanisms().get(ScramMechanism.SCRAM_SHA_256).clear());
+    }
+
     private static void testToImage(ScramImage image) {
         testToImage(image, Optional.empty());
     }