diff --git a/frameworks/kafka/src/main/dist/server.properties.mustache b/frameworks/kafka/src/main/dist/server.properties.mustache
index 5b79300d..f913158f 100644
--- a/frameworks/kafka/src/main/dist/server.properties.mustache
+++ b/frameworks/kafka/src/main/dist/server.properties.mustache
@@ -252,6 +252,8 @@ max.connections={{KAFKA_MAX_CONNECTIONS}}
 max.connections.per.ip.overrides={{KAFKA_MAX_CONNECTIONS_PER_IP_OVERRIDES}}
 max.connections.per.ip={{KAFKA_MAX_CONNECTIONS_PER_IP}}
 
+max.incremental.fetch.session.cache.slots={{KAFKA_MAX_INCREMENTAL_FETCH_SESSION_CACHE_SLOTS}}
+
 message.max.bytes={{KAFKA_MESSAGE_MAX_BYTES}}
 
 kafka.metrics.reporters={{KAFKA_METRICS_REPORTERS}}
diff --git a/frameworks/kafka/universe/config.json b/frameworks/kafka/universe/config.json
index 2870c10b..064b810c 100644
--- a/frameworks/kafka/universe/config.json
+++ b/frameworks/kafka/universe/config.json
@@ -1180,6 +1180,13 @@
           "description": "Per-ip or hostname overrides to the default maximum number of connections",
           "default": ""
         },
+        "max_incremental_fetch_session_cache_slots": {
+          "title": "max.incremental.fetch.session.cache.slots",
+          "description": "The maximum number of incremental fetch sessions that we will maintain",
+          "type": "integer",
+          "default": 1000,
+          "minimum": 0
+        },
         "num_partitions": {
           "title": "num.partitions",
           "description": "The default number of log partitions per topic",
diff --git a/frameworks/kafka/universe/marathon.json.mustache b/frameworks/kafka/universe/marathon.json.mustache
index b688c4f2..fe6cde2e 100644
--- a/frameworks/kafka/universe/marathon.json.mustache
+++ b/frameworks/kafka/universe/marathon.json.mustache
@@ -165,6 +165,7 @@
     "TASKCFG_ALL_KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS": "{{kafka.offsets_topic_num_partitions}}",
     "TASKCFG_ALL_KAFKA_MAX_CONNECTIONS": "{{kafka.max_connections}}",
     "TASKCFG_ALL_KAFKA_MAX_CONNECTIONS_PER_IP_OVERRIDES": "{{kafka.max_connections_per_ip_overrides}}",
+    "TASKCFG_ALL_KAFKA_MAX_INCREMENTAL_FETCH_SESSION_CACHE_SLOTS": "{{kafka.max_incremental_fetch_session_cache_slots}}",
     "TASKCFG_ALL_KAFKA_LEADER_IMBALANCE_CHECK_INTERVAL_SECONDS": "{{kafka.leader_imbalance_check_interval_seconds}}",
     "TASKCFG_ALL_KAFKA_INTER_BROKER_PROTOCOL_VERSION": "{{kafka.inter_broker_protocol_version}}",
     "TASKCFG_ALL_KAFKA_LOG_MESSAGE_FORMAT_VERSION": "{{kafka.log_message_format_version}}",