diff --git a/CHANGELOG.md b/CHANGELOG.md
index 456a55a64a..f5517d3960 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -51,6 +51,7 @@
 * [ENHANCEMENT] Add `ingest_storage_ingester_autoscaling_scale_up_stabilization_window_seconds` and `ingest_storage_ingester_autoscaling_scale_down_stabilization_window_seconds` config options to make stabilization window for ingester autoscaling when using ingest-storage configurable. #9445
 * [ENHANCEMENT] Make label-selector in ReplicaTemplate/ingester-zone-a object configurable when using ingest-storage. #9480
 * [ENHANCEMENT] Add `querier_only_args` option to specify CLI flags that apply only to queriers but not ruler-queriers. #9503
+* [ENHANCEMENT] Validate the Kafka client ID configured when ingest storage is enabled. #9573
 
 ### Mimirtool
 
diff --git a/operations/mimir/ingest-storage.libsonnet b/operations/mimir/ingest-storage.libsonnet
index 5517c9ceba..10c857c166 100644
--- a/operations/mimir/ingest-storage.libsonnet
+++ b/operations/mimir/ingest-storage.libsonnet
@@ -179,12 +179,30 @@
 
     if std.length(cleanSettings) > 0 then
       // Build the Kafka client ID from settings. Sort the key-value pairs to get a stable output.
-      std.join(',', std.sort(
+      local clientID = std.join(',', std.sort(
         [
           key + '=' + cleanSettings[key]
           for key in std.objectFields(cleanSettings)
         ]
-      ))
+      ));
+
+      // The client ID can be up to 255 characters in length (limit hardcoded in franz-go), and can include
+      // the following characters:
+      // a-z, A-Z, 0-9, . (dot), _ (underscore), - (dash), = (equal), "," (comma).
+      local isValid(input) =
+        std.length(
+          // Build an array of invalid characters.
+          [
+            char
+            for char in std.stringChars(clientID)
+            if !std.member('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-=,', char)
+          ]
+        ) == 0;
+
+      assert std.length(clientID) < 256 : 'the Kafka client ID must be less than 256 characters (actual: %s)' % clientID;
+      assert isValid(clientID) : 'the Kafka client ID contains invalid characters (actual: %s)' % clientID;
+
+      clientID
     else
       // Explicitly use null so that the CLI flag will not be set at all (instead of getting set to an empty string).
       null,