Repository: SeldonIO/seldon-core (language: python)

Issue:
Ability to add custom prefix to consumer ID for kafka topics in v2



Discussion / hints:

Fix patch:
diff --git a/docs/source/contents/examples/index.md b/docs/source/contents/examples/index.md
index 7a823ea8a5..c34c524c55 100644
--- a/docs/source/contents/examples/index.md
+++ b/docs/source/contents/examples/index.md
@@ -43,6 +43,10 @@ This section will provide some examples to allow operations with Seldon to be te
 
  * [Checking Pipeline readiness](pipeline-ready-and-metadata.md)
 
+## Further Kubernetes Examples
+
+ * [Kubernetes custerwide example](k8s-clusterwide.md)
+
 ## Advanced Examples
 
  * [Huggingface speech to sentiment with explanations pipeline](speech-to-sentiment.md)
diff --git a/docs/source/contents/examples/k8s-clusterwide.md b/docs/source/contents/examples/k8s-clusterwide.md
new file mode 100644
index 0000000000..5981fca689
--- /dev/null
+++ b/docs/source/contents/examples/k8s-clusterwide.md
@@ -0,0 +1,8 @@
+# Kubernetes Clusterwide Example
+
+Run these examples from the `samples` folder.
+
+
+```{include} ../../../../samples/k8s-clusterwide.md
+```
+
diff --git a/docs/source/contents/getting-started/configuration/index.md b/docs/source/contents/getting-started/configuration/index.md
index 5305d7312f..d5d45ae10c 100644
--- a/docs/source/contents/getting-started/configuration/index.md
+++ b/docs/source/contents/getting-started/configuration/index.md
@@ -13,6 +13,7 @@ We allow configuration of the Kafka integration. In general this configuration l
 The top level keys are:
 
  * `topicPrefix` : the prefix to add to kafka topics created by Seldon
+ * `consumerGroupIdPrefix` : the prefix to add to Kafka consumer group IDs created by Seldon
  * `bootstrap.servers` : the global bootstrap kafka servers to use
  * `consumer` : consumer settings
  * `producer` : producer settings
@@ -20,6 +21,8 @@ The top level keys are:
 
 For `topicPrefix` you can use any acceptable kafka topic characters which are `a-z, A-Z, 0-9, . (dot), _ (underscore), and - (dash)`. We use `.` (dot) internally as topic naming separator so we would suggest you don't end your topic prefix with a dot for clarity. For illustration, an example topic could be `seldon.default.model.mymodel.inputs` where `seldon` is the topic prefix.
 
+The `consumerGroupIdPrefix` will ensure that all consumer groups created have a given prefix.
+
 ### Kubernetes
 
 For Kubernetes this is controlled via a ConfigMap called `seldon-kafka` whose default values are defined in the `SeldonConfig` custom resource.
@@ -44,6 +47,25 @@ helm install seldon-v2-runtime k8s/helm-charts/seldon-core-v2-runtime \
     --namespace seldon-mesh \
     --values k8s/samples/values-runtime-kafka-compression.yaml
 ```
+### Topic and consumer isolation
+
+If you use a shared Kafka cluster with other applications you may want to isolate the topic names and consumer group IDs from other users of the cluster to ensure there is no name clash. For this we provide two settings:
+
+ * `topicPrefix`: set a prefix for all topics
+ * `consumerGroupIdPrefix`: set a prefix for all consumer groups
+
+An example to set this in the configuration when using the helm installation is showm below for creating the default `SeldonConfig`:
+
+```
+helm upgrade --install seldon-v2 k8s/helm-charts/seldon-core-v2-setup/ -n seldon-mesh \
+    --set controller.clusterwide=true \
+    --set kafka.topicPrefix=myorg \
+    --set kafka.consumerGroupIdPrefix=myorg
+```
+
+You can find a worked example [here](../../examples/k8s-clusterwide.md).
+
+You can create alternate `SeldonConfig`s with different values or override values for particular `SeldonRuntime` installs.
 
 ## Tracing Configuration
 
diff --git a/k8s/helm-charts/seldon-core-v2-crds/templates/seldon-v2-crds.yaml b/k8s/helm-charts/seldon-core-v2-crds/templates/seldon-v2-crds.yaml
index d2e9436a21..746ce6a020 100644
--- a/k8s/helm-charts/seldon-core-v2-crds/templates/seldon-v2-crds.yaml
+++ b/k8s/helm-charts/seldon-core-v2-crds/templates/seldon-v2-crds.yaml
@@ -8109,6 +8109,8 @@ spec:
                           - type: string
                           x-kubernetes-int-or-string: true
                         type: object
+                      consumerGroupIdPrefix:
+                        type: string
                       debug:
                         type: string
                       producer:
@@ -8220,6 +8222,8 @@ spec:
                           - type: string
                           x-kubernetes-int-or-string: true
                         type: object
+                      consumerGroupIdPrefix:
+                        type: string
                       debug:
                         type: string
                       producer:
diff --git a/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml b/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
index 0c1845b45a..46c8468324 100644
--- a/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
+++ b/k8s/helm-charts/seldon-core-v2-setup/templates/seldon-v2-components.yaml
@@ -1373,6 +1373,8 @@ spec:
       - env:
         - name: SELDON_KAFKA_BOOTSTRAP_SERVERS
           value: '{{ .Values.kafka.bootstrap }}'
+        - name: SELDON_KAFKA_CONSUMER_PREFIX
+          value: '{{ .Values.kafka.consumerGroupIdPrefix }}'
         - name: SELDON_KAFKA_REPLICATION_FACTOR
           value: '{{ .Values.kafka.topics.replicationFactor }}'
         - name: SELDON_KAFKA_PARTITIONS_DEFAULT
@@ -1423,7 +1425,7 @@ spec:
             configMapKeyRef:
               key: OTEL_EXPORTER_OTLP_ENDPOINT
               name: seldon-tracing
-        - name: POD_NAMESPACE
+        - name: SELDON_POD_NAMESPACE
           valueFrom:
             fieldRef:
               fieldPath: metadata.namespace
@@ -1459,6 +1461,7 @@ spec:
         session.timeout.ms: '{{ .Values.kafka.consumer.sessionTimeoutMs }}'
         topic.metadata.propagation.max.ms: '{{ .Values.kafka.consumer.topicMetadataPropagationMaxMs
           }}'
+      consumerGroupIdPrefix: '{{ .Values.kafka.consumerGroupIdPrefix }}'
       debug: '{{ .Values.kafka.debug }}'
       producer:
         linger.ms: '{{ .Values.kafka.producer.lingerMs }}'
diff --git a/k8s/helm-charts/seldon-core-v2-setup/values.yaml b/k8s/helm-charts/seldon-core-v2-setup/values.yaml
index 981ff8409f..306f0a1947 100644
--- a/k8s/helm-charts/seldon-core-v2-setup/values.yaml
+++ b/k8s/helm-charts/seldon-core-v2-setup/values.yaml
@@ -101,6 +101,7 @@ kafka:
   # This will need to be customized to your kafka installation broker endpoint
   bootstrap: seldon-kafka-bootstrap.seldon-mesh:9092
   topicPrefix: seldon
+  consumerGroupIdPrefix:
   consumer:
     sessionTimeoutMs: 6000
     autoOffsetReset: earliest
diff --git a/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template b/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template
index 412b79634a..36ea8f14a7 100644
--- a/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template
+++ b/k8s/helm-charts/seldon-core-v2-setup/values.yaml.template
@@ -101,6 +101,7 @@ kafka:
   # This will need to be customized to your kafka installation broker endpoint
   bootstrap: seldon-kafka-bootstrap.seldon-mesh:9092
   topicPrefix: seldon
+  consumerGroupIdPrefix:
   consumer:
     sessionTimeoutMs: 6000
     autoOffsetReset: earliest
diff --git a/k8s/kustomize/helm-components-sc/patch_dataflow.yaml b/k8s/kustomize/helm-components-sc/patch_dataflow.yaml
index 64b47daa3b..e0e7250343 100644
--- a/k8s/kustomize/helm-components-sc/patch_dataflow.yaml
+++ b/k8s/kustomize/helm-components-sc/patch_dataflow.yaml
@@ -15,6 +15,8 @@ spec:
         env:
         - name: SELDON_KAFKA_BOOTSTRAP_SERVERS
           value: '{{ .Values.kafka.bootstrap }}'
+        - name: SELDON_KAFKA_CONSUMER_PREFIX
+          value: '{{ .Values.kafka.consumerGroupIdPrefix }}'
         - name: SELDON_KAFKA_REPLICATION_FACTOR
           value: '{{ .Values.kafka.topics.replicationFactor }}'
         - name: SELDON_KAFKA_PARTITIONS_DEFAULT
diff --git a/k8s/kustomize/helm-components-sc/patch_kafkaconfig.yaml b/k8s/kustomize/helm-components-sc/patch_kafkaconfig.yaml
index 2ec3154ca1..168058ebb2 100644
--- a/k8s/kustomize/helm-components-sc/patch_kafkaconfig.yaml
+++ b/k8s/kustomize/helm-components-sc/patch_kafkaconfig.yaml
@@ -8,6 +8,7 @@ spec:
       bootstrap.servers: '{{ .Values.kafka.bootstrap }}'
       debug: '{{ .Values.kafka.debug }}'
       topicPrefix: '{{ .Values.kafka.topicPrefix }}'
+      consumerGroupIdPrefix: '{{ .Values.kafka.consumerGroupIdPrefix }}'
       consumer:
         "session.timeout.ms": '{{ .Values.kafka.consumer.sessionTimeoutMs }}'
         "auto.offset.reset": '{{ .Values.kafka.consumer.autoOffsetReset }}'
diff --git a/k8s/yaml/components.yaml b/k8s/yaml/components.yaml
index e7bd9623cb..b28cf4585d 100644
--- a/k8s/yaml/components.yaml
+++ b/k8s/yaml/components.yaml
@@ -995,6 +995,8 @@ spec:
       - env:
         - name: SELDON_KAFKA_BOOTSTRAP_SERVERS
           value: 'seldon-kafka-bootstrap.seldon-mesh:9092'
+        - name: SELDON_KAFKA_CONSUMER_PREFIX
+          value: ''
         - name: SELDON_KAFKA_REPLICATION_FACTOR
           value: '1'
         - name: SELDON_KAFKA_PARTITIONS_DEFAULT
@@ -1043,7 +1045,7 @@ spec:
             configMapKeyRef:
               key: OTEL_EXPORTER_OTLP_ENDPOINT
               name: seldon-tracing
-        - name: POD_NAMESPACE
+        - name: SELDON_POD_NAMESPACE
           valueFrom:
             fieldRef:
               fieldPath: metadata.namespace
@@ -1076,6 +1078,7 @@ spec:
         message.max.bytes: '1000000000'
         session.timeout.ms: '6000'
         topic.metadata.propagation.max.ms: '300000'
+      consumerGroupIdPrefix: ''
       debug: ''
       producer:
         linger.ms: '0'
diff --git a/k8s/yaml/crds.yaml b/k8s/yaml/crds.yaml
index 0add5813cf..34caaa1665 100644
--- a/k8s/yaml/crds.yaml
+++ b/k8s/yaml/crds.yaml
@@ -8114,6 +8114,8 @@ spec:
                           - type: string
                           x-kubernetes-int-or-string: true
                         type: object
+                      consumerGroupIdPrefix:
+                        type: string
                       debug:
                         type: string
                       producer:
@@ -8226,6 +8228,8 @@ spec:
                           - type: string
                           x-kubernetes-int-or-string: true
                         type: object
+                      consumerGroupIdPrefix:
+                        type: string
                       debug:
                         type: string
                       producer:
diff --git a/operator/apis/mlops/v1alpha1/seldonconfig_types.go b/operator/apis/mlops/v1alpha1/seldonconfig_types.go
index b3d32692fe..381fe45b03 100644
--- a/operator/apis/mlops/v1alpha1/seldonconfig_types.go
+++ b/operator/apis/mlops/v1alpha1/seldonconfig_types.go
@@ -60,12 +60,13 @@ type ServiceConfig struct {
 }
 
 type KafkaConfig struct {
-	BootstrapServers string                        `json:"bootstrap.servers,omitempty"`
-	Debug            string                        `json:"debug,omitempty"`
-	Consumer         map[string]intstr.IntOrString `json:"consumer,omitempty"`
-	Producer         map[string]intstr.IntOrString `json:"producer,omitempty"`
-	Streams          map[string]intstr.IntOrString `json:"streams,omitempty"`
-	TopicPrefix      string                        `json:"topicPrefix,omitempty"`
+	BootstrapServers      string                        `json:"bootstrap.servers,omitempty"`
+	ConsumerGroupIdPrefix string                        `json:"consumerGroupIdPrefix,omitempty"`
+	Debug                 string                        `json:"debug,omitempty"`
+	Consumer              map[string]intstr.IntOrString `json:"consumer,omitempty"`
+	Producer              map[string]intstr.IntOrString `json:"producer,omitempty"`
+	Streams               map[string]intstr.IntOrString `json:"streams,omitempty"`
+	TopicPrefix           string                        `json:"topicPrefix,omitempty"`
 }
 
 type AgentConfiguration struct {
@@ -142,6 +143,9 @@ func (k *KafkaConfig) addDefaults(defaults KafkaConfig) {
 	if k.BootstrapServers == "" {
 		k.BootstrapServers = defaults.BootstrapServers
 	}
+	if k.ConsumerGroupIdPrefix == "" {
+		k.ConsumerGroupIdPrefix = defaults.ConsumerGroupIdPrefix
+	}
 	if k.Consumer == nil && defaults.Consumer != nil {
 		k.Consumer = make(map[string]intstr.IntOrString)
 	}
diff --git a/operator/config/crd/bases/mlops.seldon.io_seldonconfigs.yaml b/operator/config/crd/bases/mlops.seldon.io_seldonconfigs.yaml
index c0c5e94d57..54f4012b53 100644
--- a/operator/config/crd/bases/mlops.seldon.io_seldonconfigs.yaml
+++ b/operator/config/crd/bases/mlops.seldon.io_seldonconfigs.yaml
@@ -7578,6 +7578,8 @@ spec:
                           - type: string
                           x-kubernetes-int-or-string: true
                         type: object
+                      consumerGroupIdPrefix:
+                        type: string
                       debug:
                         type: string
                       producer:
diff --git a/operator/config/crd/bases/mlops.seldon.io_seldonruntimes.yaml b/operator/config/crd/bases/mlops.seldon.io_seldonruntimes.yaml
index 25a0f968fd..b5f2e652e6 100644
--- a/operator/config/crd/bases/mlops.seldon.io_seldonruntimes.yaml
+++ b/operator/config/crd/bases/mlops.seldon.io_seldonruntimes.yaml
@@ -62,6 +62,8 @@ spec:
                           - type: string
                           x-kubernetes-int-or-string: true
                         type: object
+                      consumerGroupIdPrefix:
+                        type: string
                       debug:
                         type: string
                       producer:
diff --git a/operator/config/seldonconfigs/default.yaml b/operator/config/seldonconfigs/default.yaml
index 43c5693b29..a294db71f9 100644
--- a/operator/config/seldonconfigs/default.yaml
+++ b/operator/config/seldonconfigs/default.yaml
@@ -23,7 +23,7 @@ spec:
             configMapKeyRef:
               key: OTEL_EXPORTER_OTLP_ENDPOINT
               name: seldon-tracing
-        - name: POD_NAMESPACE
+        - name: SELDON_POD_NAMESPACE
           valueFrom:
             fieldRef:
               fieldPath: metadata.namespace
diff --git a/samples/k8s-clusterwide.ipynb b/samples/k8s-clusterwide.ipynb
index 56f21b0a50..7b78b46d79 100644
--- a/samples/k8s-clusterwide.ipynb
+++ b/samples/k8s-clusterwide.ipynb
@@ -11,7 +11,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 41,
+   "execution_count": 44,
    "id": "e09cb983",
    "metadata": {},
    "outputs": [
@@ -21,7 +21,7 @@
      "text": [
       "Release \"seldon-core-v2-crds\" does not exist. Installing it now.\n",
       "NAME: seldon-core-v2-crds\n",
-      "LAST DEPLOYED: Fri Jun 23 14:41:22 2023\n",
+      "LAST DEPLOYED: Tue Aug 15 11:01:03 2023\n",
       "NAMESPACE: seldon-mesh\n",
       "STATUS: deployed\n",
       "REVISION: 1\n",
@@ -33,9 +33,17 @@
     "!helm upgrade --install seldon-core-v2-crds  ../k8s/helm-charts/seldon-core-v2-crds -n seldon-mesh"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "id": "448aed9c",
+   "metadata": {},
+   "source": [
+    " The below setup also illustrates using kafka specific prefixes for topics and consumerIds for isolation where the kafka cluster is shared with other applications and you want to enforce constraints. You would not strictly need this in this example as we install Kafka just for Seldon here."
+   ]
+  },
   {
    "cell_type": "code",
-   "execution_count": 42,
+   "execution_count": 45,
    "id": "a4debc76",
    "metadata": {},
    "outputs": [
@@ -45,7 +53,7 @@
      "text": [
       "Release \"seldon-v2\" does not exist. Installing it now.\n",
       "NAME: seldon-v2\n",
-      "LAST DEPLOYED: Fri Jun 23 14:41:26 2023\n",
+      "LAST DEPLOYED: Tue Aug 15 11:01:07 2023\n",
       "NAMESPACE: seldon-mesh\n",
       "STATUS: deployed\n",
       "REVISION: 1\n",
@@ -54,12 +62,15 @@
     }
    ],
    "source": [
-    "!helm upgrade --install seldon-v2 ../k8s/helm-charts/seldon-core-v2-setup/ -n seldon-mesh --set controller.clusterwide=true"
+    "!helm upgrade --install seldon-v2 ../k8s/helm-charts/seldon-core-v2-setup/ -n seldon-mesh \\\n",
+    "    --set controller.clusterwide=true \\\n",
+    "    --set kafka.topicPrefix=myorg \\\n",
+    "    --set kafka.consumerGroupIdPrefix=myorg"
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": 43,
+   "execution_count": 46,
    "id": "29353104",
    "metadata": {},
    "outputs": [
@@ -79,7 +90,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 10,
+   "execution_count": 47,
    "id": "56cdf114",
    "metadata": {},
    "outputs": [
@@ -88,7 +99,7 @@
      "output_type": "stream",
      "text": [
       "NAME: seldon-v2-runtime\r\n",
-      "LAST DEPLOYED: Fri Jun 23 14:10:23 2023\r\n",
+      "LAST DEPLOYED: Tue Aug 15 11:01:11 2023\r\n",
       "NAMESPACE: ns1\r\n",
       "STATUS: deployed\r\n",
       "REVISION: 1\r\n",
@@ -102,7 +113,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 11,
+   "execution_count": 14,
    "id": "4b815db6",
    "metadata": {},
    "outputs": [
@@ -111,7 +122,7 @@
      "output_type": "stream",
      "text": [
       "NAME: seldon-v2-servers\r\n",
-      "LAST DEPLOYED: Fri Jun 23 14:10:38 2023\r\n",
+      "LAST DEPLOYED: Tue Aug 15 10:47:31 2023\r\n",
       "NAMESPACE: ns1\r\n",
       "STATUS: deployed\r\n",
       "REVISION: 1\r\n",
@@ -125,7 +136,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 12,
+   "execution_count": 15,
    "id": "3ce2d189",
    "metadata": {},
    "outputs": [
@@ -134,7 +145,7 @@
      "output_type": "stream",
      "text": [
       "NAME: seldon-v2-runtime\r\n",
-      "LAST DEPLOYED: Fri Jun 23 14:10:42 2023\r\n",
+      "LAST DEPLOYED: Tue Aug 15 10:53:12 2023\r\n",
       "NAMESPACE: ns2\r\n",
       "STATUS: deployed\r\n",
       "REVISION: 1\r\n",
@@ -148,7 +159,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 13,
+   "execution_count": 16,
    "id": "4a214689",
    "metadata": {},
    "outputs": [
@@ -157,7 +168,7 @@
      "output_type": "stream",
      "text": [
       "NAME: seldon-v2-servers\r\n",
-      "LAST DEPLOYED: Fri Jun 23 14:10:44 2023\r\n",
+      "LAST DEPLOYED: Tue Aug 15 10:53:28 2023\r\n",
       "NAMESPACE: ns2\r\n",
       "STATUS: deployed\r\n",
       "REVISION: 1\r\n",
@@ -171,17 +182,55 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 14,
+   "execution_count": 17,
+   "id": "b194c88e",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "server.mlops.seldon.io/mlserver condition met\n",
+      "server.mlops.seldon.io/triton condition met\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl wait --for condition=ready --timeout=300s server --all -n ns1"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 18,
+   "id": "60016845",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "server.mlops.seldon.io/mlserver condition met\n",
+      "server.mlops.seldon.io/triton condition met\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl wait --for condition=ready --timeout=300s server --all -n ns2"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 19,
    "id": "sunrise-commercial",
    "metadata": {},
    "outputs": [
     {
      "data": {
       "text/plain": [
-       "'172.21.255.2'"
+       "'172.18.255.2'"
       ]
      },
-     "execution_count": 14,
+     "execution_count": 19,
      "metadata": {},
      "output_type": "execute_result"
     }
@@ -196,17 +245,17 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 15,
+   "execution_count": 20,
    "id": "012886cc",
    "metadata": {},
    "outputs": [
     {
      "data": {
       "text/plain": [
-       "'172.21.255.4'"
+       "'172.18.255.4'"
       ]
      },
-     "execution_count": 15,
+     "execution_count": 20,
      "metadata": {},
      "output_type": "execute_result"
     }
@@ -229,7 +278,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 16,
+   "execution_count": 21,
    "id": "operating-console",
    "metadata": {},
    "outputs": [
@@ -255,7 +304,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 17,
+   "execution_count": 22,
    "id": "exempt-bumper",
    "metadata": {
     "scrolled": true
@@ -275,7 +324,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 18,
+   "execution_count": 23,
    "id": "beneficial-logan",
    "metadata": {},
    "outputs": [
@@ -293,7 +342,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 19,
+   "execution_count": 24,
    "id": "67900afd",
    "metadata": {},
    "outputs": [
@@ -304,7 +353,7 @@
       "{\r\n",
       "\t\"model_name\": \"iris_1\",\r\n",
       "\t\"model_version\": \"1\",\r\n",
-      "\t\"id\": \"276de7e7-9f2f-4329-9179-08d4d54bb0b5\",\r\n",
+      "\t\"id\": \"3ca1757c-df02-4e57-87c1-38311bcc5943\",\r\n",
       "\t\"parameters\": {},\r\n",
       "\t\"outputs\": [\r\n",
       "\t\t{\r\n",
@@ -333,7 +382,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 20,
+   "execution_count": 25,
    "id": "a1848c7d",
    "metadata": {},
    "outputs": [
@@ -375,7 +424,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 21,
+   "execution_count": 26,
    "id": "0843b13e",
    "metadata": {
     "scrolled": true
@@ -395,7 +444,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 22,
+   "execution_count": 27,
    "id": "99e270be",
    "metadata": {},
    "outputs": [
@@ -413,7 +462,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 23,
+   "execution_count": 28,
    "id": "ae262b3d",
    "metadata": {},
    "outputs": [
@@ -424,7 +473,7 @@
       "{\r\n",
       "\t\"model_name\": \"iris_1\",\r\n",
       "\t\"model_version\": \"1\",\r\n",
-      "\t\"id\": \"7c0bb004-be2c-4483-97a9-6fd2e0a834ad\",\r\n",
+      "\t\"id\": \"f706a23e-775f-4765-bd18-2e98d83bf7d5\",\r\n",
       "\t\"parameters\": {},\r\n",
       "\t\"outputs\": [\r\n",
       "\t\t{\r\n",
@@ -453,7 +502,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 24,
+   "execution_count": 29,
    "id": "30b3342f",
    "metadata": {},
    "outputs": [
@@ -495,7 +544,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 25,
+   "execution_count": 30,
    "id": "precious-development",
    "metadata": {},
    "outputs": [
@@ -513,6 +562,377 @@
     "!kubectl delete -f ./models/sklearn-iris-gs.yaml -n ns2"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "id": "0b091f23",
+   "metadata": {},
+   "source": [
+    "## Pipelines"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 31,
+   "id": "21705cff",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "model.mlops.seldon.io/tfsimple1 created\n",
+      "model.mlops.seldon.io/tfsimple2 created\n",
+      "model.mlops.seldon.io/tfsimple1 created\n",
+      "model.mlops.seldon.io/tfsimple2 created\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl create -f ./models/tfsimple1.yaml -n ns1\n",
+    "!kubectl create -f ./models/tfsimple2.yaml -n ns1\n",
+    "!kubectl create -f ./models/tfsimple1.yaml -n ns2\n",
+    "!kubectl create -f ./models/tfsimple2.yaml -n ns2"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 32,
+   "id": "277f1ff7",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "model.mlops.seldon.io/tfsimple1 condition met\n",
+      "model.mlops.seldon.io/tfsimple2 condition met\n",
+      "model.mlops.seldon.io/tfsimple1 condition met\n",
+      "model.mlops.seldon.io/tfsimple2 condition met\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl wait --for condition=ready --timeout=300s model --all -n ns1\n",
+    "!kubectl wait --for condition=ready --timeout=300s model --all -n ns2"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 33,
+   "id": "733b3f95",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "pipeline.mlops.seldon.io/tfsimples created\n",
+      "pipeline.mlops.seldon.io/tfsimples created\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl create -f ./pipelines/tfsimples.yaml -n ns1\n",
+    "!kubectl create -f ./pipelines/tfsimples.yaml -n ns2"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 34,
+   "id": "16dcb6e3",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "pipeline.mlops.seldon.io/tfsimples condition met\n",
+      "pipeline.mlops.seldon.io/tfsimples condition met\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl wait --for condition=ready --timeout=300s pipeline --all -n ns1\n",
+    "!kubectl wait --for condition=ready --timeout=300s pipeline --all -n ns2"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 35,
+   "id": "7b2f86fc",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "{\r\n",
+      "  \"outputs\": [\r\n",
+      "    {\r\n",
+      "      \"name\": \"OUTPUT0\",\r\n",
+      "      \"datatype\": \"INT32\",\r\n",
+      "      \"shape\": [\r\n",
+      "        \"1\",\r\n",
+      "        \"16\"\r\n",
+      "      ],\r\n",
+      "      \"contents\": {\r\n",
+      "        \"intContents\": [\r\n",
+      "          2,\r\n",
+      "          4,\r\n",
+      "          6,\r\n",
+      "          8,\r\n",
+      "          10,\r\n",
+      "          12,\r\n",
+      "          14,\r\n",
+      "          16,\r\n",
+      "          18,\r\n",
+      "          20,\r\n",
+      "          22,\r\n",
+      "          24,\r\n",
+      "          26,\r\n",
+      "          28,\r\n",
+      "          30,\r\n",
+      "          32\r\n",
+      "        ]\r\n",
+      "      }\r\n",
+      "    },\r\n",
+      "    {\r\n",
+      "      \"name\": \"OUTPUT1\",\r\n",
+      "      \"datatype\": \"INT32\",\r\n",
+      "      \"shape\": [\r\n",
+      "        \"1\",\r\n",
+      "        \"16\"\r\n",
+      "      ],\r\n",
+      "      \"contents\": {\r\n",
+      "        \"intContents\": [\r\n",
+      "          2,\r\n",
+      "          4,\r\n",
+      "          6,\r\n",
+      "          8,\r\n",
+      "          10,\r\n",
+      "          12,\r\n",
+      "          14,\r\n",
+      "          16,\r\n",
+      "          18,\r\n",
+      "          20,\r\n",
+      "          22,\r\n",
+      "          24,\r\n",
+      "          26,\r\n",
+      "          28,\r\n",
+      "          30,\r\n",
+      "          32\r\n",
+      "        ]\r\n",
+      "      }\r\n",
+      "    }\r\n",
+      "  ]\r\n",
+      "}\r\n"
+     ]
+    }
+   ],
+   "source": [
+    "!seldon pipeline infer tfsimples --inference-mode grpc --inference-host ${MESH_IP_NS1}:80 \\\n",
+    "    '{\"model_name\":\"simple\",\"inputs\":[{\"name\":\"INPUT0\",\"contents\":{\"int_contents\":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},\"datatype\":\"INT32\",\"shape\":[1,16]},{\"name\":\"INPUT1\",\"contents\":{\"int_contents\":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},\"datatype\":\"INT32\",\"shape\":[1,16]}]}' | jq -M ."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 36,
+   "id": "9529a039",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "{\r\n",
+      "  \"outputs\": [\r\n",
+      "    {\r\n",
+      "      \"name\": \"OUTPUT0\",\r\n",
+      "      \"datatype\": \"INT32\",\r\n",
+      "      \"shape\": [\r\n",
+      "        \"1\",\r\n",
+      "        \"16\"\r\n",
+      "      ],\r\n",
+      "      \"contents\": {\r\n",
+      "        \"intContents\": [\r\n",
+      "          2,\r\n",
+      "          4,\r\n",
+      "          6,\r\n",
+      "          8,\r\n",
+      "          10,\r\n",
+      "          12,\r\n",
+      "          14,\r\n",
+      "          16,\r\n",
+      "          18,\r\n",
+      "          20,\r\n",
+      "          22,\r\n",
+      "          24,\r\n",
+      "          26,\r\n",
+      "          28,\r\n",
+      "          30,\r\n",
+      "          32\r\n",
+      "        ]\r\n",
+      "      }\r\n",
+      "    },\r\n",
+      "    {\r\n",
+      "      \"name\": \"OUTPUT1\",\r\n",
+      "      \"datatype\": \"INT32\",\r\n",
+      "      \"shape\": [\r\n",
+      "        \"1\",\r\n",
+      "        \"16\"\r\n",
+      "      ],\r\n",
+      "      \"contents\": {\r\n",
+      "        \"intContents\": [\r\n",
+      "          2,\r\n",
+      "          4,\r\n",
+      "          6,\r\n",
+      "          8,\r\n",
+      "          10,\r\n",
+      "          12,\r\n",
+      "          14,\r\n",
+      "          16,\r\n",
+      "          18,\r\n",
+      "          20,\r\n",
+      "          22,\r\n",
+      "          24,\r\n",
+      "          26,\r\n",
+      "          28,\r\n",
+      "          30,\r\n",
+      "          32\r\n",
+      "        ]\r\n",
+      "      }\r\n",
+      "    }\r\n",
+      "  ]\r\n",
+      "}\r\n"
+     ]
+    }
+   ],
+   "source": [
+    "!seldon pipeline infer tfsimples --inference-mode grpc --inference-host ${MESH_IP_NS2}:80 \\\n",
+    "    '{\"model_name\":\"simple\",\"inputs\":[{\"name\":\"INPUT0\",\"contents\":{\"int_contents\":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},\"datatype\":\"INT32\",\"shape\":[1,16]},{\"name\":\"INPUT1\",\"contents\":{\"int_contents\":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},\"datatype\":\"INT32\",\"shape\":[1,16]}]}' | jq -M ."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "a3fe4c34",
+   "metadata": {},
+   "source": [
+    "If you have installed Kafka via the ansible playbook setup-ecosystem then you can use the following command to see the consumer group ids which are reflecting the settings we created."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 49,
+   "id": "6a1eb5f7",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "myorg-ns2-seldon-pipelinegateway-dfd61b49-4bb9-4684-adce-0b7cc215d3af\r\n",
+      "myorg-ns2-seldon-modelgateway-17\r\n",
+      "myorg-ns1-seldon-pipelinegateway-d4fc83e6-29cb-442e-90cd-92a389961cfe\r\n",
+      "myorg-ns2-seldon-modelgateway-60\r\n",
+      "myorg-ns2-seldon-dataflow-73d465744b7b1b5be20e88d6245e50bd\r\n",
+      "myorg-ns1-seldon-modelgateway-60\r\n",
+      "myorg-ns1-seldon-modelgateway-17\r\n",
+      "myorg-ns1-seldon-dataflow-f563e04e093caa20c03e6eced084331b\r\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl exec seldon-kafka-0 -n seldon-mesh -- bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "8a360fcb",
+   "metadata": {},
+   "source": [
+    "We can similarly show the topics that have been created."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 50,
+   "id": "71285ba0",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "__consumer_offsets\r\n",
+      "myorg.ns1.errors.errors\r\n",
+      "myorg.ns1.model.iris.inputs\r\n",
+      "myorg.ns1.model.iris.outputs\r\n",
+      "myorg.ns1.model.tfsimple1.inputs\r\n",
+      "myorg.ns1.model.tfsimple1.outputs\r\n",
+      "myorg.ns1.model.tfsimple2.inputs\r\n",
+      "myorg.ns1.model.tfsimple2.outputs\r\n",
+      "myorg.ns1.pipeline.tfsimples.inputs\r\n",
+      "myorg.ns1.pipeline.tfsimples.outputs\r\n",
+      "myorg.ns2.errors.errors\r\n",
+      "myorg.ns2.model.iris.inputs\r\n",
+      "myorg.ns2.model.iris.outputs\r\n",
+      "myorg.ns2.model.tfsimple1.inputs\r\n",
+      "myorg.ns2.model.tfsimple1.outputs\r\n",
+      "myorg.ns2.model.tfsimple2.inputs\r\n",
+      "myorg.ns2.model.tfsimple2.outputs\r\n",
+      "myorg.ns2.pipeline.tfsimples.inputs\r\n",
+      "myorg.ns2.pipeline.tfsimples.outputs\r\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl exec seldon-kafka-0 -n seldon-mesh -- bin/kafka-topics.sh --bootstrap-server=localhost:9092 --list"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 37,
+   "id": "b0b4b82f",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "pipeline.mlops.seldon.io \"tfsimples\" deleted\n",
+      "pipeline.mlops.seldon.io \"tfsimples\" deleted\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl delete -f ./pipelines/tfsimples.yaml -n ns1\n",
+    "!kubectl delete -f ./pipelines/tfsimples.yaml -n ns2"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 38,
+   "id": "941ed2c3",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "model.mlops.seldon.io \"tfsimple1\" deleted\n",
+      "model.mlops.seldon.io \"tfsimple2\" deleted\n",
+      "model.mlops.seldon.io \"tfsimple1\" deleted\n",
+      "model.mlops.seldon.io \"tfsimple2\" deleted\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl delete -f ./models/tfsimple1.yaml -n ns1\n",
+    "!kubectl delete -f ./models/tfsimple2.yaml -n ns1\n",
+    "!kubectl delete -f ./models/tfsimple1.yaml -n ns2\n",
+    "!kubectl delete -f ./models/tfsimple2.yaml -n ns2"
+   ]
+  },
   {
    "cell_type": "markdown",
    "id": "280c1dcd",
@@ -523,7 +943,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 26,
+   "execution_count": 39,
    "id": "e694edee",
    "metadata": {},
    "outputs": [
@@ -543,7 +963,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 35,
+   "execution_count": 40,
    "id": "8b00bd3e",
    "metadata": {},
    "outputs": [
@@ -552,7 +972,7 @@
      "output_type": "stream",
      "text": [
       "release \"seldon-v2-runtime\" uninstalled\n",
-      "Error: uninstall: Release not loaded: seldon-v2-runtime: release: not found\n"
+      "release \"seldon-v2-runtime\" uninstalled\n"
      ]
     }
    ],
@@ -563,7 +983,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 38,
+   "execution_count": 41,
    "id": "4c671429",
    "metadata": {},
    "outputs": [
@@ -581,7 +1001,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 39,
+   "execution_count": 42,
    "id": "72b24840",
    "metadata": {},
    "outputs": [
@@ -599,7 +1019,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 40,
+   "execution_count": 43,
    "id": "e7773481",
    "metadata": {},
    "outputs": [
diff --git a/samples/k8s-clusterwide.md b/samples/k8s-clusterwide.md
index 0877de4833..4a586c3fb9 100644
--- a/samples/k8s-clusterwide.md
+++ b/samples/k8s-clusterwide.md
@@ -7,7 +7,7 @@ helm upgrade --install seldon-core-v2-crds  ../k8s/helm-charts/seldon-core-v2-cr
 ```
 Release "seldon-core-v2-crds" does not exist. Installing it now.
 NAME: seldon-core-v2-crds
-LAST DEPLOYED: Fri Jun 23 14:41:22 2023
+LAST DEPLOYED: Tue Aug 15 11:01:03 2023
 NAMESPACE: seldon-mesh
 STATUS: deployed
 REVISION: 1
@@ -15,14 +15,19 @@ TEST SUITE: None
 
 ```
 
+ The below setup also illustrates using kafka specific prefixes for topics and consumerIds for isolation where the kafka cluster is shared with other applications and you want to enforce constraints. You would not strictly need this in this example as we install Kafka just for Seldon here.
+
 ```bash
-helm upgrade --install seldon-v2 ../k8s/helm-charts/seldon-core-v2-setup/ -n seldon-mesh --set controller.clusterwide=true
+helm upgrade --install seldon-v2 ../k8s/helm-charts/seldon-core-v2-setup/ -n seldon-mesh \
+    --set controller.clusterwide=true \
+    --set kafka.topicPrefix=myorg \
+    --set kafka.consumerGroupIdPrefix=myorg
 ```
 
 ```
 Release "seldon-v2" does not exist. Installing it now.
 NAME: seldon-v2
-LAST DEPLOYED: Fri Jun 23 14:41:26 2023
+LAST DEPLOYED: Tue Aug 15 11:01:07 2023
 NAMESPACE: seldon-mesh
 STATUS: deployed
 REVISION: 1
@@ -47,7 +52,7 @@ helm install seldon-v2-runtime ../k8s/helm-charts/seldon-core-v2-runtime  -n ns1
 
 ```yaml
 NAME: seldon-v2-runtime
-LAST DEPLOYED: Fri Jun 23 14:10:23 2023
+LAST DEPLOYED: Tue Aug 15 11:01:11 2023
 NAMESPACE: ns1
 STATUS: deployed
 REVISION: 1
@@ -61,7 +66,7 @@ helm install seldon-v2-servers ../k8s/helm-charts/seldon-core-v2-servers  -n ns1
 
 ```yaml
 NAME: seldon-v2-servers
-LAST DEPLOYED: Fri Jun 23 14:10:38 2023
+LAST DEPLOYED: Tue Aug 15 10:47:31 2023
 NAMESPACE: ns1
 STATUS: deployed
 REVISION: 1
@@ -75,7 +80,7 @@ helm install seldon-v2-runtime ../k8s/helm-charts/seldon-core-v2-runtime  -n ns2
 
 ```yaml
 NAME: seldon-v2-runtime
-LAST DEPLOYED: Fri Jun 23 14:10:42 2023
+LAST DEPLOYED: Tue Aug 15 10:53:12 2023
 NAMESPACE: ns2
 STATUS: deployed
 REVISION: 1
@@ -89,7 +94,7 @@ helm install seldon-v2-servers ../k8s/helm-charts/seldon-core-v2-servers  -n ns2
 
 ```yaml
 NAME: seldon-v2-servers
-LAST DEPLOYED: Fri Jun 23 14:10:44 2023
+LAST DEPLOYED: Tue Aug 15 10:53:28 2023
 NAMESPACE: ns2
 STATUS: deployed
 REVISION: 1
@@ -97,6 +102,26 @@ TEST SUITE: None
 
 ```
 
+```bash
+kubectl wait --for condition=ready --timeout=300s server --all -n ns1
+```
+
+```
+server.mlops.seldon.io/mlserver condition met
+server.mlops.seldon.io/triton condition met
+
+```
+
+```bash
+kubectl wait --for condition=ready --timeout=300s server --all -n ns2
+```
+
+```
+server.mlops.seldon.io/mlserver condition met
+server.mlops.seldon.io/triton condition met
+
+```
+
 ```python
 MESH_IP=!kubectl get svc seldon-mesh -n ns1 -o jsonpath='{.status.loadBalancer.ingress[0].ip}'
 MESH_IP_NS1=MESH_IP[0]
@@ -106,7 +131,7 @@ MESH_IP_NS1
 ```
 
 ```
-'172.21.255.2'
+'172.18.255.2'
 
 ```
 
@@ -119,7 +144,7 @@ MESH_IP_NS2
 ```
 
 ```
-'172.21.255.4'
+'172.18.255.4'
 
 ```
 
@@ -169,7 +194,7 @@ seldon model infer iris --inference-host ${MESH_IP_NS1}:80 \
 {
 	"model_name": "iris_1",
 	"model_version": "1",
-	"id": "276de7e7-9f2f-4329-9179-08d4d54bb0b5",
+	"id": "3ca1757c-df02-4e57-87c1-38311bcc5943",
 	"parameters": {},
 	"outputs": [
 		{
@@ -251,7 +276,7 @@ seldon model infer iris --inference-host ${MESH_IP_NS2}:80 \
 {
 	"model_name": "iris_1",
 	"model_version": "1",
-	"id": "7c0bb004-be2c-4483-97a9-6fd2e0a834ad",
+	"id": "f706a23e-775f-4765-bd18-2e98d83bf7d5",
 	"parameters": {},
 	"outputs": [
 		{
@@ -317,6 +342,269 @@ model.mlops.seldon.io "iris" deleted
 
 ```
 
+## Pipelines
+
+```bash
+kubectl create -f ./models/tfsimple1.yaml -n ns1
+kubectl create -f ./models/tfsimple2.yaml -n ns1
+kubectl create -f ./models/tfsimple1.yaml -n ns2
+kubectl create -f ./models/tfsimple2.yaml -n ns2
+```
+
+```
+model.mlops.seldon.io/tfsimple1 created
+model.mlops.seldon.io/tfsimple2 created
+model.mlops.seldon.io/tfsimple1 created
+model.mlops.seldon.io/tfsimple2 created
+
+```
+
+```bash
+kubectl wait --for condition=ready --timeout=300s model --all -n ns1
+kubectl wait --for condition=ready --timeout=300s model --all -n ns2
+```
+
+```
+model.mlops.seldon.io/tfsimple1 condition met
+model.mlops.seldon.io/tfsimple2 condition met
+model.mlops.seldon.io/tfsimple1 condition met
+model.mlops.seldon.io/tfsimple2 condition met
+
+```
+
+```bash
+kubectl create -f ./pipelines/tfsimples.yaml -n ns1
+kubectl create -f ./pipelines/tfsimples.yaml -n ns2
+```
+
+```
+pipeline.mlops.seldon.io/tfsimples created
+pipeline.mlops.seldon.io/tfsimples created
+
+```
+
+```bash
+kubectl wait --for condition=ready --timeout=300s pipeline --all -n ns1
+kubectl wait --for condition=ready --timeout=300s pipeline --all -n ns2
+```
+
+```
+pipeline.mlops.seldon.io/tfsimples condition met
+pipeline.mlops.seldon.io/tfsimples condition met
+
+```
+
+```bash
+seldon pipeline infer tfsimples --inference-mode grpc --inference-host ${MESH_IP_NS1}:80 \
+    '{"model_name":"simple","inputs":[{"name":"INPUT0","contents":{"int_contents":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},"datatype":"INT32","shape":[1,16]},{"name":"INPUT1","contents":{"int_contents":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},"datatype":"INT32","shape":[1,16]}]}' | jq -M .
+```
+
+```json
+{
+  "outputs": [
+    {
+      "name": "OUTPUT0",
+      "datatype": "INT32",
+      "shape": [
+        "1",
+        "16"
+      ],
+      "contents": {
+        "intContents": [
+          2,
+          4,
+          6,
+          8,
+          10,
+          12,
+          14,
+          16,
+          18,
+          20,
+          22,
+          24,
+          26,
+          28,
+          30,
+          32
+        ]
+      }
+    },
+    {
+      "name": "OUTPUT1",
+      "datatype": "INT32",
+      "shape": [
+        "1",
+        "16"
+      ],
+      "contents": {
+        "intContents": [
+          2,
+          4,
+          6,
+          8,
+          10,
+          12,
+          14,
+          16,
+          18,
+          20,
+          22,
+          24,
+          26,
+          28,
+          30,
+          32
+        ]
+      }
+    }
+  ]
+}
+
+```
+
+```bash
+seldon pipeline infer tfsimples --inference-mode grpc --inference-host ${MESH_IP_NS2}:80 \
+    '{"model_name":"simple","inputs":[{"name":"INPUT0","contents":{"int_contents":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},"datatype":"INT32","shape":[1,16]},{"name":"INPUT1","contents":{"int_contents":[1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]},"datatype":"INT32","shape":[1,16]}]}' | jq -M .
+```
+
+```json
+{
+  "outputs": [
+    {
+      "name": "OUTPUT0",
+      "datatype": "INT32",
+      "shape": [
+        "1",
+        "16"
+      ],
+      "contents": {
+        "intContents": [
+          2,
+          4,
+          6,
+          8,
+          10,
+          12,
+          14,
+          16,
+          18,
+          20,
+          22,
+          24,
+          26,
+          28,
+          30,
+          32
+        ]
+      }
+    },
+    {
+      "name": "OUTPUT1",
+      "datatype": "INT32",
+      "shape": [
+        "1",
+        "16"
+      ],
+      "contents": {
+        "intContents": [
+          2,
+          4,
+          6,
+          8,
+          10,
+          12,
+          14,
+          16,
+          18,
+          20,
+          22,
+          24,
+          26,
+          28,
+          30,
+          32
+        ]
+      }
+    }
+  ]
+}
+
+```
+
+If you have installed Kafka via the ansible playbook setup-ecosystem then you can use the following command to see the consumer group ids which are reflecting the settings we created.
+
+```bash
+kubectl exec seldon-kafka-0 -n seldon-mesh -- bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092
+```
+
+```
+myorg-ns2-seldon-pipelinegateway-dfd61b49-4bb9-4684-adce-0b7cc215d3af
+myorg-ns2-seldon-modelgateway-17
+myorg-ns1-seldon-pipelinegateway-d4fc83e6-29cb-442e-90cd-92a389961cfe
+myorg-ns2-seldon-modelgateway-60
+myorg-ns2-seldon-dataflow-73d465744b7b1b5be20e88d6245e50bd
+myorg-ns1-seldon-modelgateway-60
+myorg-ns1-seldon-modelgateway-17
+myorg-ns1-seldon-dataflow-f563e04e093caa20c03e6eced084331b
+
+```
+
+We can similarly show the topics that have been created.
+
+```bash
+kubectl exec seldon-kafka-0 -n seldon-mesh -- bin/kafka-topics.sh --bootstrap-server=localhost:9092 --list
+```
+
+```
+__consumer_offsets
+myorg.ns1.errors.errors
+myorg.ns1.model.iris.inputs
+myorg.ns1.model.iris.outputs
+myorg.ns1.model.tfsimple1.inputs
+myorg.ns1.model.tfsimple1.outputs
+myorg.ns1.model.tfsimple2.inputs
+myorg.ns1.model.tfsimple2.outputs
+myorg.ns1.pipeline.tfsimples.inputs
+myorg.ns1.pipeline.tfsimples.outputs
+myorg.ns2.errors.errors
+myorg.ns2.model.iris.inputs
+myorg.ns2.model.iris.outputs
+myorg.ns2.model.tfsimple1.inputs
+myorg.ns2.model.tfsimple1.outputs
+myorg.ns2.model.tfsimple2.inputs
+myorg.ns2.model.tfsimple2.outputs
+myorg.ns2.pipeline.tfsimples.inputs
+myorg.ns2.pipeline.tfsimples.outputs
+
+```
+
+```bash
+kubectl delete -f ./pipelines/tfsimples.yaml -n ns1
+kubectl delete -f ./pipelines/tfsimples.yaml -n ns2
+```
+
+```
+pipeline.mlops.seldon.io "tfsimples" deleted
+pipeline.mlops.seldon.io "tfsimples" deleted
+
+```
+
+```bash
+kubectl delete -f ./models/tfsimple1.yaml -n ns1
+kubectl delete -f ./models/tfsimple2.yaml -n ns1
+kubectl delete -f ./models/tfsimple1.yaml -n ns2
+kubectl delete -f ./models/tfsimple2.yaml -n ns2
+```
+
+```
+model.mlops.seldon.io "tfsimple1" deleted
+model.mlops.seldon.io "tfsimple2" deleted
+model.mlops.seldon.io "tfsimple1" deleted
+model.mlops.seldon.io "tfsimple2" deleted
+
+```
+
 ## TearDown
 
 ```bash
@@ -337,7 +625,7 @@ helm delete seldon-v2-runtime -n ns2 --wait
 
 ```
 release "seldon-v2-runtime" uninstalled
-Error: uninstall: Release not loaded: seldon-v2-runtime: release: not found
+release "seldon-v2-runtime" uninstalled
 
 ```
 
diff --git a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/Cli.kt b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/Cli.kt
index 23f6b030cf..24d76817b7 100644
--- a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/Cli.kt
+++ b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/Cli.kt
@@ -29,6 +29,7 @@ object Cli {
     // General setup
     val logLevelApplication = Key("log.level.app", enumType(*Level.values()))
     val logLevelKafka = Key("log.level.kafka", enumType(*Level.values()))
+    val namespace = Key("pod.namespace", stringType)
 
     // Seldon components
     val upstreamHost = Key("upstream.host", stringType)
@@ -36,6 +37,7 @@ object Cli {
 
     // Kafka
     val kafkaBootstrapServers = Key("kafka.bootstrap.servers", stringType)
+    val kafkaConsumerGroupIdPrefix = Key("kafka.consumer.prefix", stringType)
     val kafkaSecurityProtocol = Key("kafka.security.protocol", enumType(*KafkaSecurityProtocols))
     val kafkaPartitions = Key("kafka.partitions.default", intType)
     val kafkaReplicationFactor = Key("kafka.replication.factor", intType)
@@ -62,9 +64,11 @@ object Cli {
         return listOf(
             logLevelApplication,
             logLevelKafka,
+            namespace,
             upstreamHost,
             upstreamPort,
             kafkaBootstrapServers,
+            kafkaConsumerGroupIdPrefix,
             kafkaSecurityProtocol,
             kafkaPartitions,
             kafkaReplicationFactor,
diff --git a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/Main.kt b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/Main.kt
index ad6ce0329c..f3d5194a00 100644
--- a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/Main.kt
+++ b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/Main.kt
@@ -83,6 +83,8 @@ object Main {
             config[Cli.upstreamHost],
             config[Cli.upstreamPort],
             GrpcServiceConfigProvider.config,
+            config[Cli.kafkaConsumerGroupIdPrefix],
+            config[Cli.namespace],
         )
 
         addShutdownHandler(subscriber)
diff --git a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/PipelineSubscriber.kt b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/PipelineSubscriber.kt
index a6fff2fdb8..16f3e84c9f 100644
--- a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/PipelineSubscriber.kt
+++ b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/PipelineSubscriber.kt
@@ -49,6 +49,8 @@ class PipelineSubscriber(
     private val upstreamHost: String,
     private val upstreamPort: Int,
     grpcServiceConfig: Map<String, Any>,
+    private val kafkaConsumerGroupIdPrefix: String,
+    private val namespace: String,
 ) {
     private val kafkaAdmin = KafkaAdmin(kafkaAdminProperties, kafkaStreamsParams)
     private val channel = ManagedChannelBuilder
@@ -72,7 +74,7 @@ class PipelineSubscriber(
     suspend fun subscribe() {
         logger.info("will connect to ${upstreamHost}:${upstreamPort}")
         retry(grpcFailurePolicy + binaryExponentialBackoff(50..5_000L)) {
-            subscribePipelines()
+            subscribePipelines(kafkaConsumerGroupIdPrefix, namespace)
         }
     }
 
@@ -83,7 +85,7 @@ class PipelineSubscriber(
     //  Pipeline UID should be enough to uniquely key it, even across versions?
     //  ...
     //  - Add map of model name -> (weak) referrents/reference count to avoid recreation of streams
-    private suspend fun subscribePipelines() {
+    private suspend fun subscribePipelines(kafkaConsumerGroupIdPrefix: String, namespace: String) {
         logger.info("Subscribing to pipeline updates")
         client
             .subscribePipelineUpdates(request = makeSubscriptionRequest())
@@ -97,7 +99,7 @@ class PipelineSubscriber(
                 )
 
                 when (update.op) {
-                    PipelineOperation.Create -> handleCreate(metadata, update.updatesList)
+                    PipelineOperation.Create -> handleCreate(metadata, update.updatesList, kafkaConsumerGroupIdPrefix, namespace)
                     PipelineOperation.Delete -> handleDelete(metadata)
                     else -> logger.warn("unrecognised pipeline operation (${update.op})")
                 }
@@ -119,9 +121,11 @@ class PipelineSubscriber(
     private suspend fun handleCreate(
         metadata: PipelineMetadata,
         steps: List<PipelineStepUpdate>,
+        kafkaConsumerGroupIdPrefix: String,
+        namespace: String,
     ) {
         logger.info("Create pipeline ${metadata.name} version: ${metadata.version} id: ${metadata.id}")
-        val pipeline = Pipeline.forSteps(metadata, steps, kafkaProperties, kafkaDomainParams)
+        val pipeline = Pipeline.forSteps(metadata, steps, kafkaProperties, kafkaDomainParams, kafkaConsumerGroupIdPrefix, namespace)
         if (pipeline.size != steps.size) {
             client.pipelineUpdateEvent(
                 makePipelineUpdateEvent(
diff --git a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Configuration.kt b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Configuration.kt
index 5cf833b952..3c12cc398d 100644
--- a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Configuration.kt
+++ b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Configuration.kt
@@ -156,11 +156,20 @@ fun getKafkaProperties(params: KafkaStreamsParams): KafkaProperties {
     }
 }
 
-fun KafkaProperties.withAppId(name: String): KafkaProperties {
+fun KafkaProperties.withAppId(namespace: String, consumerGroupIdPrefix: String, name: String): KafkaProperties {
     val properties = KafkaProperties()
 
     properties.putAll(this.toMap())
-    properties[StreamsConfig.APPLICATION_ID_CONFIG] = "seldon-dataflow-$name"
+    val str = StringBuilder()
+    if (consumerGroupIdPrefix.isNotEmpty()) {
+        str.append(consumerGroupIdPrefix+"-")
+    }
+    if (namespace.isNotEmpty()) {
+        str.append(namespace+"-")
+    }
+    str.append("seldon-dataflow-"+name)
+    properties[StreamsConfig.APPLICATION_ID_CONFIG] = str.toString()
+
     // TODO add k8s host name to ensure static membership is only used for consumers from the same pod restarting?
     //
     // If set, allows static membership which would allow restarts within SESSION_TIMEOUT_MS_CONFIG
diff --git a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Pipeline.kt b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Pipeline.kt
index 818422f808..f0b52aece2 100644
--- a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Pipeline.kt
+++ b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Pipeline.kt
@@ -83,9 +83,11 @@ class Pipeline(
             steps: List<PipelineStepUpdate>,
             kafkaProperties: KafkaProperties,
             kafkaDomainParams: KafkaDomainParams,
+            kafkaConsumerGroupIdPrefix: String,
+            namespace: String,
         ): Pipeline {
             val (topology, numSteps) = buildTopology(metadata, steps, kafkaDomainParams)
-            val pipelineProperties = localiseKafkaProperties(kafkaProperties, metadata, numSteps)
+            val pipelineProperties = localiseKafkaProperties(kafkaProperties, metadata, numSteps, kafkaConsumerGroupIdPrefix, namespace)
             val streamsApp = KafkaStreams(topology, pipelineProperties)
             logger.info("Create pipeline stream for name:${metadata.name} id:${metadata.id} version:${metadata.version} stream with kstream app id:${pipelineProperties[StreamsConfig.APPLICATION_ID_CONFIG]}")
             return Pipeline(metadata, topology, streamsApp, kafkaDomainParams, numSteps)
@@ -120,9 +122,13 @@ class Pipeline(
             kafkaProperties: KafkaProperties,
             metadata: PipelineMetadata,
             numSteps: Int,
+            kafkaConsumerGroupIdPrefix: String,
+            namespace: String
         ): KafkaProperties {
             return kafkaProperties
                 .withAppId(
+                    namespace,
+                    kafkaConsumerGroupIdPrefix,
                     HashUtils.hashIfLong(metadata.id),
                 )
                 .withStreamThreads(
diff --git a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/mtls/K8sCertSecretsProvider.kt b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/mtls/K8sCertSecretsProvider.kt
index 970120002a..169260a9c0 100644
--- a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/mtls/K8sCertSecretsProvider.kt
+++ b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/mtls/K8sCertSecretsProvider.kt
@@ -31,7 +31,7 @@ import java.nio.file.Files
 object K8sCertSecretsProvider {
 
     private val kubeConfigPath: String = System.getenv("HOME") + "/.kube/config"
-    private val namespace = System.getenv("POD_NAMESPACE")
+    private val namespace = System.getenv("SELDON_POD_NAMESPACE")
 
     private fun getApiClient(): ApiClient = try {
         ClientBuilder.cluster().build()
diff --git a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/sasl/K8sPasswordSecretsProvider.kt b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/sasl/K8sPasswordSecretsProvider.kt
index b34e4d5959..25c109e595 100644
--- a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/sasl/K8sPasswordSecretsProvider.kt
+++ b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/sasl/K8sPasswordSecretsProvider.kt
@@ -29,7 +29,7 @@ import java.io.FileReader
 
 object K8sPasswordSecretsProvider {
     private val kubeConfigPath: String = System.getenv("HOME") + "/.kube/config"
-    private val namespace = System.getenv("POD_NAMESPACE")
+    private val namespace = System.getenv("SELDON_POD_NAMESPACE")
 
     private fun getApiClient(): ApiClient = try {
         ClientBuilder.cluster().build()
diff --git a/scheduler/data-flow/src/main/resources/local.properties b/scheduler/data-flow/src/main/resources/local.properties
index 9f5fedc124..c0e3a69985 100644
--- a/scheduler/data-flow/src/main/resources/local.properties
+++ b/scheduler/data-flow/src/main/resources/local.properties
@@ -1,6 +1,7 @@
 log.level.app=INFO
 log.level.kafka=WARN
 kafka.bootstrap.servers=localhost:9092
+kafka.consumer.prefix=
 kafka.security.protocol=PLAINTEXT
 kafka.partitions.default=1
 kafka.replication.factor=1
diff --git a/scheduler/pkg/kafka/config/config.go b/scheduler/pkg/kafka/config/config.go
index a03b96a4bd..9f5b19fc59 100644
--- a/scheduler/pkg/kafka/config/config.go
+++ b/scheduler/pkg/kafka/config/config.go
@@ -25,12 +25,13 @@ import (
 )
 
 type KafkaConfig struct {
-	BootstrapServers string          `json:"bootstrap.servers,omitempty"`
-	Debug            string          `json:"debug,omitempty"`
-	Consumer         kafka.ConfigMap `json:"consumer,omitempty"`
-	Producer         kafka.ConfigMap `json:"producer,omitempty"`
-	Streams          kafka.ConfigMap `json:"streams,omitempty"`
-	TopicPrefix      string          `json:"topicPrefix,omitempty"`
+	BootstrapServers      string          `json:"bootstrap.servers,omitempty"`
+	Debug                 string          `json:"debug,omitempty"`
+	Consumer              kafka.ConfigMap `json:"consumer,omitempty"`
+	Producer              kafka.ConfigMap `json:"producer,omitempty"`
+	Streams               kafka.ConfigMap `json:"streams,omitempty"`
+	TopicPrefix           string          `json:"topicPrefix,omitempty"`
+	ConsumerGroupIdPrefix string          `json:"consumerGroupIdPrefix,omitempty"`
 }
 
 type none = struct{}
diff --git a/scheduler/pkg/kafka/gateway/manager.go b/scheduler/pkg/kafka/gateway/manager.go
index e779423f09..21d91e1f5d 100644
--- a/scheduler/pkg/kafka/gateway/manager.go
+++ b/scheduler/pkg/kafka/gateway/manager.go
@@ -120,7 +120,12 @@ func (cm *ConsumerManager) createKafkaConfigs(kafkaConfig *ManagerConfig) error
 func (cm *ConsumerManager) getInferKafkaConsumer(modelName string, create bool) (*InferKafkaHandler, error) {
 	logger := cm.logger.WithField("func", "getInferKafkaConsumer")
 
-	consumerBucketId := util.GetKafkaConsumerName(modelName, modelGatewayConsumerNamePrefix, cm.maxNumConsumers)
+	consumerBucketId := util.GetKafkaConsumerName(
+		cm.managerConfig.Namespace,
+		cm.managerConfig.SeldonKafkaConfig.ConsumerGroupIdPrefix,
+		modelName,
+		modelGatewayConsumerNamePrefix,
+		cm.maxNumConsumers)
 	ic, ok := cm.consumers[consumerBucketId]
 	logger.Debugf("Getting consumer for model %s with bucket id %s", modelName, consumerBucketId)
 	if !ok && !create {
diff --git a/scheduler/pkg/kafka/pipeline/consumer_manager.go b/scheduler/pkg/kafka/pipeline/consumer_manager.go
index 30ea6425cb..bc19b91c37 100644
--- a/scheduler/pkg/kafka/pipeline/consumer_manager.go
+++ b/scheduler/pkg/kafka/pipeline/consumer_manager.go
@@ -18,6 +18,7 @@ package pipeline
 
 import (
 	"fmt"
+	"strings"
 	"sync"
 
 	"github.com/google/uuid"
@@ -44,9 +45,11 @@ type ConsumerManager struct {
 	maxNumConsumers         int
 	maxNumTopicsPerConsumer int
 	tracer                  trace.Tracer
+	namespace               string
 }
 
 func NewConsumerManager(
+	namespace string,
 	logger log.FieldLogger,
 	consumerConfig *config.KafkaConfig,
 	maxNumTopicsPerConsumer,
@@ -59,6 +62,7 @@ func NewConsumerManager(
 		Info("creating consumer manager")
 
 	return &ConsumerManager{
+		namespace:               namespace,
 		logger:                  logger.WithField("source", "ConsumerManager"),
 		consumerConfig:          consumerConfig,
 		maxNumTopicsPerConsumer: maxNumTopicsPerConsumer,
@@ -75,7 +79,7 @@ func (cm *ConsumerManager) createConsumer() error {
 	c, err := NewMultiTopicsKafkaConsumer(
 		cm.logger,
 		cm.consumerConfig,
-		getKafkaConsumerName(kafkaConsumerNamePrefix, uuid.New().String()),
+		getKafkaConsumerName(cm.namespace, cm.consumerConfig.ConsumerGroupIdPrefix, kafkaConsumerNamePrefix, uuid.New().String()),
 		cm.tracer,
 	)
 	if err != nil {
@@ -132,6 +136,14 @@ func (cm *ConsumerManager) Stop() {
 	}
 }
 
-func getKafkaConsumerName(prefix, id string) string {
-	return fmt.Sprintf("%s-%s", prefix, id)
+func getKafkaConsumerName(namespace, consumerGroupIdPrefix, componentPrefix, id string) string {
+	var sb strings.Builder
+	if consumerGroupIdPrefix != "" {
+		sb.WriteString(consumerGroupIdPrefix + "-")
+	}
+	if namespace != "" {
+		sb.WriteString(namespace + "-")
+	}
+	sb.WriteString(componentPrefix + "-" + id)
+	return sb.String()
 }
diff --git a/scheduler/pkg/kafka/pipeline/kafkamanager.go b/scheduler/pkg/kafka/pipeline/kafkamanager.go
index 8d80e9d2db..81ed8de10a 100644
--- a/scheduler/pkg/kafka/pipeline/kafkamanager.go
+++ b/scheduler/pkg/kafka/pipeline/kafkamanager.go
@@ -98,7 +98,7 @@ func NewKafkaManager(
 		logger:          logger.WithField("source", "KafkaManager"),
 		topicNamer:      topicNamer,
 		tracer:          tracer,
-		consumerManager: NewConsumerManager(logger, kafkaConfig, maxNumTopicsPerConsumer, maxNumConsumers, tracer),
+		consumerManager: NewConsumerManager(namespace, logger, kafkaConfig, maxNumTopicsPerConsumer, maxNumConsumers, tracer),
 		mu:              sync.RWMutex{},
 	}
 
diff --git a/scheduler/pkg/util/hash.go b/scheduler/pkg/util/hash.go
index 377a7d8eff..534da11d07 100644
--- a/scheduler/pkg/util/hash.go
+++ b/scheduler/pkg/util/hash.go
@@ -19,6 +19,7 @@ package util
 import (
 	"fmt"
 	"hash/fnv"
+	"strings"
 
 	"github.com/OneOfOne/xxhash"
 )
@@ -50,7 +51,16 @@ func modelIdToConsumerBucket(modelId string, numBuckets int) uint32 {
 	return hash % uint32(numBuckets)
 }
 
-func GetKafkaConsumerName(modelName, prefix string, maxConsumers int) string {
+func GetKafkaConsumerName(namespace, consumerGroupIdPrefix, modelName, prefix string, maxConsumers int) string {
 	idx := modelIdToConsumerBucket(modelName, maxConsumers)
-	return fmt.Sprintf("%s-%d", prefix, idx)
+	var sb strings.Builder
+	if consumerGroupIdPrefix != "" {
+		sb.WriteString(consumerGroupIdPrefix + "-")
+	}
+	if namespace != "" {
+		sb.WriteString(namespace + "-")
+	}
+	sb.WriteString(prefix + "-")
+	sb.WriteString(fmt.Sprintf("%d", idx))
+	return sb.String()
 }


Test patch:
diff --git a/operator/apis/mlops/v1alpha1/seldonconfig_types_test.go b/operator/apis/mlops/v1alpha1/seldonconfig_types_test.go
index 0c6c7e10fc..29c95c6a6f 100644
--- a/operator/apis/mlops/v1alpha1/seldonconfig_types_test.go
+++ b/operator/apis/mlops/v1alpha1/seldonconfig_types_test.go
@@ -79,7 +79,8 @@ func TestSeldonConfigurationAddDefaults(t *testing.T) {
 			name: "kafka overrides",
 			defaults: SeldonConfiguration{
 				KafkaConfig: KafkaConfig{
-					BootstrapServers: "h1,h2",
+					BootstrapServers:      "h1,h2",
+					ConsumerGroupIdPrefix: "foo",
 					Consumer: map[string]intstr.IntOrString{
 						"key1": intstr.FromInt(1000),
 					},
@@ -93,7 +94,8 @@ func TestSeldonConfigurationAddDefaults(t *testing.T) {
 			},
 			runtime: SeldonConfiguration{
 				KafkaConfig: KafkaConfig{
-					Consumer: map[string]intstr.IntOrString{},
+					ConsumerGroupIdPrefix: "bar",
+					Consumer:              map[string]intstr.IntOrString{},
 					Producer: map[string]intstr.IntOrString{
 						"key4": intstr.FromString("val"),
 					},
@@ -104,7 +106,8 @@ func TestSeldonConfigurationAddDefaults(t *testing.T) {
 			},
 			expected: SeldonConfiguration{
 				KafkaConfig: KafkaConfig{
-					BootstrapServers: "h1,h2",
+					BootstrapServers:      "h1,h2",
+					ConsumerGroupIdPrefix: "bar",
 					Consumer: map[string]intstr.IntOrString{
 						"key1": intstr.FromInt(1000),
 					},
@@ -122,7 +125,8 @@ func TestSeldonConfigurationAddDefaults(t *testing.T) {
 			name: "kafka no overrides",
 			defaults: SeldonConfiguration{
 				KafkaConfig: KafkaConfig{
-					BootstrapServers: "h1,h2",
+					BootstrapServers:      "h1,h2",
+					ConsumerGroupIdPrefix: "foo",
 					Consumer: map[string]intstr.IntOrString{
 						"key1": intstr.FromInt(1000),
 					},
@@ -139,7 +143,8 @@ func TestSeldonConfigurationAddDefaults(t *testing.T) {
 			},
 			expected: SeldonConfiguration{
 				KafkaConfig: KafkaConfig{
-					BootstrapServers: "h1,h2",
+					BootstrapServers:      "h1,h2",
+					ConsumerGroupIdPrefix: "foo",
 					Consumer: map[string]intstr.IntOrString{
 						"key1": intstr.FromInt(1000),
 					},
diff --git a/scheduler/pkg/kafka/pipeline/consumer_manager_test.go b/scheduler/pkg/kafka/pipeline/consumer_manager_test.go
new file mode 100644
index 0000000000..9360079978
--- /dev/null
+++ b/scheduler/pkg/kafka/pipeline/consumer_manager_test.go
@@ -0,0 +1,65 @@
+package pipeline
+
+import (
+	"testing"
+
+	. "github.com/onsi/gomega"
+)
+
+func TestGetKafkaConsumerName(t *testing.T) {
+	g := NewGomegaWithT(t)
+
+	type test struct {
+		name                  string
+		namespace             string
+		consumerGroupIdPrefix string
+		componentPrefix       string
+		id                    string
+		expected              string
+	}
+	tests := []test{
+		{
+			name:                  "all params no namespace",
+			namespace:             "",
+			consumerGroupIdPrefix: "foo",
+			componentPrefix:       "pipeline",
+			id:                    "id",
+			expected:              "foo-pipeline-id",
+		},
+		{
+			name:                  "no consumer group prefix no namespace",
+			namespace:             "",
+			consumerGroupIdPrefix: "",
+			componentPrefix:       "pipeline",
+			id:                    "id",
+			expected:              "pipeline-id",
+		},
+		{
+			name:                  "all params",
+			namespace:             "default",
+			consumerGroupIdPrefix: "foo",
+			componentPrefix:       "pipeline",
+			id:                    "id",
+			expected:              "foo-default-pipeline-id",
+		},
+		{
+			name:                  "no consumer group prefix",
+			namespace:             "default",
+			consumerGroupIdPrefix: "",
+			componentPrefix:       "pipeline",
+			id:                    "id",
+			expected:              "default-pipeline-id",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			g.Expect(
+				getKafkaConsumerName(
+					test.namespace, test.consumerGroupIdPrefix, test.componentPrefix, test.id),
+			).To(Equal(
+				test.expected),
+			)
+		})
+	}
+}
diff --git a/scheduler/pkg/util/hash_test.go b/scheduler/pkg/util/hash_test.go
index ceca65481e..bb7c212f62 100644
--- a/scheduler/pkg/util/hash_test.go
+++ b/scheduler/pkg/util/hash_test.go
@@ -56,3 +56,64 @@ func TestConsistentHash(t *testing.T) {
 		})
 	}
 }
+
+func TestGetKafkaConsumerName(t *testing.T) {
+	g := NewGomegaWithT(t)
+
+	type test struct {
+		name                  string
+		namespace             string
+		consumerGroupIdPrefix string
+		modelName             string
+		prefix                string
+		maxConsumers          int
+		expected              string
+	}
+	tests := []test{
+		{
+			name:                  "all params except namespace",
+			consumerGroupIdPrefix: "foo",
+			modelName:             "model",
+			prefix:                "p",
+			maxConsumers:          1,
+			expected:              "foo-p-0",
+		},
+		{
+			name:                  "no consumer prefix or namespace",
+			consumerGroupIdPrefix: "",
+			modelName:             "model",
+			prefix:                "p",
+			maxConsumers:          1,
+			expected:              "p-0",
+		},
+		{
+			name:                  "all params",
+			namespace:             "default",
+			consumerGroupIdPrefix: "foo",
+			modelName:             "model",
+			prefix:                "p",
+			maxConsumers:          1,
+			expected:              "foo-default-p-0",
+		},
+		{
+			name:                  "no consumer prefix",
+			namespace:             "default",
+			consumerGroupIdPrefix: "",
+			modelName:             "model",
+			prefix:                "p",
+			maxConsumers:          1,
+			expected:              "default-p-0",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			g.Expect(
+				GetKafkaConsumerName(test.namespace, test.consumerGroupIdPrefix, test.modelName, test.prefix, test.maxConsumers),
+			).To(Equal(
+				test.expected,
+			))
+
+		})
+	}
+}