Repository: argoproj-labs/old-argo-dataflow (language: python)

Issue:
Test with large messages (>1Mb)
| Source | |
|---|---|
| HTTP | | 
| Kafka | | 
| STAN | |

| Sink | |
|---|---|
| HTTP | | 
| Kafka | | 
| STAN | |


Discussion / hints:
Running 1Mb messages locally caused Docker for Desktop to crash. 100Kb is ok.
Tested in GKE
Command: ` env MESSAGE_SIZE=10000000000 make test-http-stress`

| Source | 1 Mb|1.5 Mb|2 Mb|2.5 Mb|3 Mb|3.5 Mb|5 Mb|10 Mb|50 Mb|100 Mb|1 GB|
|---|---|---|---|---|---|---|---|---|---|---|---|
| HTTP |TPS 1181 |TPS 1126|TPS 1122|TPS 1136| TPS 1137|TPS 1126| TPS 1139| TPS 1182| TPS 1182|TPS 1178|TPS 1132|
| Kafka |1606 |1702|1902|1674|1440|1742|1899|1439|1607|1477|1413|
| STAN |1320 |1228|1291|1283|1241|1274|1250|1337|1209|1289|207|1379|

| Source | 1 Mb|1.5 Mb|2 Mb|2.5 Mb|3 Mb|3.5 Mb|5 Mb|10Mb|50 Mb|100 Mb|1 GB|
|---|---|---|---|---|---|---|---|---|---|---|---|
| HTTP |TPS 775 |TPS 783|TPS 782|TPS 795| TPS 795|TPS 783| TPS 769|TPS 750|TPS 769|TPS 783|TPS 776|
| Kafka | 768| 923|855|862|905|1018|912|949|864|842|886|
| STAN |208|220|215|210|212|222|223|211|211|218|207|223|
@alexec Do you have any other datapoints?


I'm not sure about these numbers. I would expect TPS to become smaller and message size grows. But that does not appear to be the case?
@alexec Same thing I discussed with Derek. I am checking message size kafka and STAN.  I am checking

Fix patch:
diff --git a/api/v1alpha1/abstract_step.go b/api/v1alpha1/abstract_step.go
new file mode 100644
index 00000000..cabc15f1
--- /dev/null
+++ b/api/v1alpha1/abstract_step.go
@@ -0,0 +1,8 @@
+package v1alpha1
+
+import corev1 "k8s.io/api/core/v1"
+
+type AbstractStep struct {
+	// +kubebuilder:default={limits: {"cpu": "500m", "memory": "256Mi"}, requests: {"cpu": "100m", "memory": "64Mi"}}
+	Resources corev1.ResourceRequirements `json:"resources,omitempty" protobuf:"bytes,1,opt,name=resources"`
+}
diff --git a/api/v1alpha1/cat.go b/api/v1alpha1/cat.go
index 8d6062df..b12d4628 100644
--- a/api/v1alpha1/cat.go
+++ b/api/v1alpha1/cat.go
@@ -4,11 +4,18 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
-type Cat struct{}
+type Cat struct {
+	AbstractStep `json:",inline" protobuf:"bytes,1,opt,name=abstractStep"`
+}
 
 func (m Cat) getContainer(req getContainerReq) corev1.Container {
+	resources := standardResources
+	if m.Resources.Size() > 0 {
+		resources = m.Resources
+	}
 	return containerBuilder{}.
 		init(req).
 		args("cat").
+		resources(resources).
 		build()
 }
diff --git a/api/v1alpha1/generated.proto b/api/v1alpha1/generated.proto
index 16d9ecc7..a0624060 100644
--- a/api/v1alpha1/generated.proto
+++ b/api/v1alpha1/generated.proto
@@ -15,7 +15,7 @@ limitations under the License.
 */
 // This file was autogenerated by go-to-protobuf. Do not edit it manually!
 
-syntax = "proto2";
+syntax = 'proto2';
 
 package github.com.argoproj_labs.argo_dataflow.api.v1alpha1;
 
@@ -38,6 +38,11 @@ message AWSEndpoint {
   optional string url = 1;
 }
 
+message AbstractStep {
+  // +kubebuilder:default={limits: {"cpu": "500m", "memory": "256Mi"}, requests: {"cpu": "100m", "memory": "64Mi"}}
+  optional k8s.io.api.core.v1.ResourceRequirements resources = 1;
+}
+
 message Backoff {
   // +kubebuilder:default="100ms"
   optional k8s.io.apimachinery.pkg.apis.meta.v1.Duration duration = 4;
@@ -58,6 +63,7 @@ message Backoff {
 }
 
 message Cat {
+  optional AbstractStep abstractStep = 1;
 }
 
 message Code {
@@ -256,6 +262,8 @@ message KafkaConfig {
   optional string version = 2;
 
   optional KafkaNET net = 3;
+
+  optional int32 maxMessageBytes = 4;
 }
 
 message KafkaNET {
diff --git a/api/v1alpha1/kafka.go b/api/v1alpha1/kafka.go
index 6eff4446..b77254b4 100644
--- a/api/v1alpha1/kafka.go
+++ b/api/v1alpha1/kafka.go
@@ -6,9 +6,10 @@ type KafkaNET struct {
 }
 
 type KafkaConfig struct {
-	Brokers []string  `json:"brokers,omitempty" protobuf:"bytes,1,rep,name=brokers"`
-	Version string    `json:"version,omitempty" protobuf:"bytes,2,opt,name=version"`
-	NET     *KafkaNET `json:"net,omitempty" protobuf:"bytes,3,opt,name=net"`
+	Brokers         []string  `json:"brokers,omitempty" protobuf:"bytes,1,rep,name=brokers"`
+	Version         string    `json:"version,omitempty" protobuf:"bytes,2,opt,name=version"`
+	NET             *KafkaNET `json:"net,omitempty" protobuf:"bytes,3,opt,name=net"`
+	MaxMessageBytes int32     `json:"maxMessageBytes,omitempty" protobuf:"varint,4,opt,name=maxMessageBytes"`
 }
 
 type Kafka struct {
diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go
index e4429af3..1f432708 100644
--- a/api/v1alpha1/zz_generated.deepcopy.go
+++ b/api/v1alpha1/zz_generated.deepcopy.go
@@ -59,6 +59,22 @@ func (in *AWSEndpoint) DeepCopy() *AWSEndpoint {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *AbstractStep) DeepCopyInto(out *AbstractStep) {
+	*out = *in
+	in.Resources.DeepCopyInto(&out.Resources)
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AbstractStep.
+func (in *AbstractStep) DeepCopy() *AbstractStep {
+	if in == nil {
+		return nil
+	}
+	out := new(AbstractStep)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Backoff) DeepCopyInto(out *Backoff) {
 	*out = *in
@@ -87,6 +103,7 @@ func (in *Backoff) DeepCopy() *Backoff {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Cat) DeepCopyInto(out *Cat) {
 	*out = *in
+	in.AbstractStep.DeepCopyInto(&out.AbstractStep)
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Cat.
@@ -1259,7 +1276,7 @@ func (in *StepSpec) DeepCopyInto(out *StepSpec) {
 	if in.Cat != nil {
 		in, out := &in.Cat, &out.Cat
 		*out = new(Cat)
-		**out = **in
+		(*in).DeepCopyInto(*out)
 	}
 	if in.Container != nil {
 		in, out := &in.Container, &out.Container
diff --git a/config/apps/nats/single-server-nats.yml b/config/apps/nats/single-server-nats.yml
index 7ca9d3e2..6143fc6b 100644
--- a/config/apps/nats/single-server-nats.yml
+++ b/config/apps/nats/single-server-nats.yml
@@ -6,6 +6,7 @@ metadata:
   name: nats-config
 data:
   nats.conf: |
+    max_payload: 50000000
     pid_file: "/var/run/nats/nats.pid"
     http: 8222
     authorization {
diff --git a/config/apps/stan.yaml b/config/apps/stan.yaml
index 3d06a8c4..ebbaecce 100644
--- a/config/apps/stan.yaml
+++ b/config/apps/stan.yaml
@@ -1,6 +1,7 @@
 apiVersion: v1
 data:
   nats.conf: |
+    max_payload: 50000000
     pid_file: "/var/run/nats/nats.pid"
     http: 8222
     authorization {
@@ -16,6 +17,7 @@ data:
   stan.conf: |
     port: 4222
     http: 8222
+    max_payload: 50000000
     authorization {
       token: "testingtokentestingtoken"
     }
diff --git a/config/apps/stan/single-server-stan.yml b/config/apps/stan/single-server-stan.yml
index 1b26723c..594c0aab 100644
--- a/config/apps/stan/single-server-stan.yml
+++ b/config/apps/stan/single-server-stan.yml
@@ -8,6 +8,7 @@ data:
   stan.conf: |
     port: 4222
     http: 8222
+    max_payload: 50000000
     authorization {
       token: "testingtokentestingtoken"
     }
diff --git a/config/ci.yaml b/config/ci.yaml
index 800380e1..05a6759a 100644
--- a/config/ci.yaml
+++ b/config/ci.yaml
@@ -681,6 +681,42 @@ spec:
                           type: object
                       type: object
                     cat:
+                      properties:
+                        resources:
+                          default:
+                            limits:
+                              cpu: 500m
+                              memory: 256Mi
+                            requests:
+                              cpu: 100m
+                              memory: 64Mi
+                          description: ResourceRequirements describes the compute
+                            resource requirements.
+                          properties:
+                            limits:
+                              additionalProperties:
+                                anyOf:
+                                - type: integer
+                                - type: string
+                                pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                x-kubernetes-int-or-string: true
+                              description: 'Limits describes the maximum amount of
+                                compute resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                              type: object
+                            requests:
+                              additionalProperties:
+                                anyOf:
+                                - type: integer
+                                - type: string
+                                pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                x-kubernetes-int-or-string: true
+                              description: 'Requests describes the minimum amount
+                                of compute resources required. If Requests is omitted
+                                for a container, it defaults to Limits if that is
+                                explicitly specified, otherwise to an implementation-defined
+                                value. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                              type: object
+                          type: object
                       type: object
                     code:
                       properties:
@@ -1386,6 +1422,9 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              maxMessageBytes:
+                                format: int32
+                                type: integer
                               name:
                                 default: default
                                 type: string
@@ -3327,6 +3366,9 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              maxMessageBytes:
+                                format: int32
+                                type: integer
                               name:
                                 default: default
                                 type: string
@@ -7543,6 +7585,41 @@ spec:
                     type: object
                 type: object
               cat:
+                properties:
+                  resources:
+                    default:
+                      limits:
+                        cpu: 500m
+                        memory: 256Mi
+                      requests:
+                        cpu: 100m
+                        memory: 64Mi
+                    description: ResourceRequirements describes the compute resource
+                      requirements.
+                    properties:
+                      limits:
+                        additionalProperties:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
+                        description: 'Limits describes the maximum amount of compute
+                          resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                        type: object
+                      requests:
+                        additionalProperties:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
+                        description: 'Requests describes the minimum amount of compute
+                          resources required. If Requests is omitted for a container,
+                          it defaults to Limits if that is explicitly specified, otherwise
+                          to an implementation-defined value. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                        type: object
+                    type: object
                 type: object
               code:
                 properties:
@@ -8233,6 +8310,9 @@ spec:
                           items:
                             type: string
                           type: array
+                        maxMessageBytes:
+                          format: int32
+                          type: integer
                         name:
                           default: default
                           type: string
@@ -10060,6 +10140,9 @@ spec:
                           items:
                             type: string
                           type: array
+                        maxMessageBytes:
+                          format: int32
+                          type: integer
                         name:
                           default: default
                           type: string
diff --git a/config/crd/bases/dataflow.argoproj.io_pipelines.yaml b/config/crd/bases/dataflow.argoproj.io_pipelines.yaml
index 76d319e0..3ed3ea20 100644
--- a/config/crd/bases/dataflow.argoproj.io_pipelines.yaml
+++ b/config/crd/bases/dataflow.argoproj.io_pipelines.yaml
@@ -676,6 +676,42 @@ spec:
                           type: object
                       type: object
                     cat:
+                      properties:
+                        resources:
+                          default:
+                            limits:
+                              cpu: 500m
+                              memory: 256Mi
+                            requests:
+                              cpu: 100m
+                              memory: 64Mi
+                          description: ResourceRequirements describes the compute
+                            resource requirements.
+                          properties:
+                            limits:
+                              additionalProperties:
+                                anyOf:
+                                - type: integer
+                                - type: string
+                                pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                x-kubernetes-int-or-string: true
+                              description: 'Limits describes the maximum amount of
+                                compute resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                              type: object
+                            requests:
+                              additionalProperties:
+                                anyOf:
+                                - type: integer
+                                - type: string
+                                pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                x-kubernetes-int-or-string: true
+                              description: 'Requests describes the minimum amount
+                                of compute resources required. If Requests is omitted
+                                for a container, it defaults to Limits if that is
+                                explicitly specified, otherwise to an implementation-defined
+                                value. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                              type: object
+                          type: object
                       type: object
                     code:
                       properties:
@@ -1381,6 +1417,9 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              maxMessageBytes:
+                                format: int32
+                                type: integer
                               name:
                                 default: default
                                 type: string
@@ -3322,6 +3361,9 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              maxMessageBytes:
+                                format: int32
+                                type: integer
                               name:
                                 default: default
                                 type: string
diff --git a/config/crd/bases/dataflow.argoproj.io_steps.yaml b/config/crd/bases/dataflow.argoproj.io_steps.yaml
index 8d893837..16471102 100644
--- a/config/crd/bases/dataflow.argoproj.io_steps.yaml
+++ b/config/crd/bases/dataflow.argoproj.io_steps.yaml
@@ -642,6 +642,41 @@ spec:
                     type: object
                 type: object
               cat:
+                properties:
+                  resources:
+                    default:
+                      limits:
+                        cpu: 500m
+                        memory: 256Mi
+                      requests:
+                        cpu: 100m
+                        memory: 64Mi
+                    description: ResourceRequirements describes the compute resource
+                      requirements.
+                    properties:
+                      limits:
+                        additionalProperties:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
+                        description: 'Limits describes the maximum amount of compute
+                          resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                        type: object
+                      requests:
+                        additionalProperties:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
+                        description: 'Requests describes the minimum amount of compute
+                          resources required. If Requests is omitted for a container,
+                          it defaults to Limits if that is explicitly specified, otherwise
+                          to an implementation-defined value. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                        type: object
+                    type: object
                 type: object
               code:
                 properties:
@@ -1332,6 +1367,9 @@ spec:
                           items:
                             type: string
                           type: array
+                        maxMessageBytes:
+                          format: int32
+                          type: integer
                         name:
                           default: default
                           type: string
@@ -3159,6 +3197,9 @@ spec:
                           items:
                             type: string
                           type: array
+                        maxMessageBytes:
+                          format: int32
+                          type: integer
                         name:
                           default: default
                           type: string
diff --git a/config/default.yaml b/config/default.yaml
index 952a3acc..a27969f3 100644
--- a/config/default.yaml
+++ b/config/default.yaml
@@ -681,6 +681,42 @@ spec:
                           type: object
                       type: object
                     cat:
+                      properties:
+                        resources:
+                          default:
+                            limits:
+                              cpu: 500m
+                              memory: 256Mi
+                            requests:
+                              cpu: 100m
+                              memory: 64Mi
+                          description: ResourceRequirements describes the compute
+                            resource requirements.
+                          properties:
+                            limits:
+                              additionalProperties:
+                                anyOf:
+                                - type: integer
+                                - type: string
+                                pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                x-kubernetes-int-or-string: true
+                              description: 'Limits describes the maximum amount of
+                                compute resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                              type: object
+                            requests:
+                              additionalProperties:
+                                anyOf:
+                                - type: integer
+                                - type: string
+                                pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                x-kubernetes-int-or-string: true
+                              description: 'Requests describes the minimum amount
+                                of compute resources required. If Requests is omitted
+                                for a container, it defaults to Limits if that is
+                                explicitly specified, otherwise to an implementation-defined
+                                value. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                              type: object
+                          type: object
                       type: object
                     code:
                       properties:
@@ -1386,6 +1422,9 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              maxMessageBytes:
+                                format: int32
+                                type: integer
                               name:
                                 default: default
                                 type: string
@@ -3327,6 +3366,9 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              maxMessageBytes:
+                                format: int32
+                                type: integer
                               name:
                                 default: default
                                 type: string
@@ -7543,6 +7585,41 @@ spec:
                     type: object
                 type: object
               cat:
+                properties:
+                  resources:
+                    default:
+                      limits:
+                        cpu: 500m
+                        memory: 256Mi
+                      requests:
+                        cpu: 100m
+                        memory: 64Mi
+                    description: ResourceRequirements describes the compute resource
+                      requirements.
+                    properties:
+                      limits:
+                        additionalProperties:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
+                        description: 'Limits describes the maximum amount of compute
+                          resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                        type: object
+                      requests:
+                        additionalProperties:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
+                        description: 'Requests describes the minimum amount of compute
+                          resources required. If Requests is omitted for a container,
+                          it defaults to Limits if that is explicitly specified, otherwise
+                          to an implementation-defined value. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                        type: object
+                    type: object
                 type: object
               code:
                 properties:
@@ -8233,6 +8310,9 @@ spec:
                           items:
                             type: string
                           type: array
+                        maxMessageBytes:
+                          format: int32
+                          type: integer
                         name:
                           default: default
                           type: string
@@ -10060,6 +10140,9 @@ spec:
                           items:
                             type: string
                           type: array
+                        maxMessageBytes:
+                          format: int32
+                          type: integer
                         name:
                           default: default
                           type: string
diff --git a/config/dev.yaml b/config/dev.yaml
index 25eb55d7..e17a48f7 100644
--- a/config/dev.yaml
+++ b/config/dev.yaml
@@ -681,6 +681,42 @@ spec:
                           type: object
                       type: object
                     cat:
+                      properties:
+                        resources:
+                          default:
+                            limits:
+                              cpu: 500m
+                              memory: 256Mi
+                            requests:
+                              cpu: 100m
+                              memory: 64Mi
+                          description: ResourceRequirements describes the compute
+                            resource requirements.
+                          properties:
+                            limits:
+                              additionalProperties:
+                                anyOf:
+                                - type: integer
+                                - type: string
+                                pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                x-kubernetes-int-or-string: true
+                              description: 'Limits describes the maximum amount of
+                                compute resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                              type: object
+                            requests:
+                              additionalProperties:
+                                anyOf:
+                                - type: integer
+                                - type: string
+                                pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                x-kubernetes-int-or-string: true
+                              description: 'Requests describes the minimum amount
+                                of compute resources required. If Requests is omitted
+                                for a container, it defaults to Limits if that is
+                                explicitly specified, otherwise to an implementation-defined
+                                value. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                              type: object
+                          type: object
                       type: object
                     code:
                       properties:
@@ -1386,6 +1422,9 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              maxMessageBytes:
+                                format: int32
+                                type: integer
                               name:
                                 default: default
                                 type: string
@@ -3327,6 +3366,9 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              maxMessageBytes:
+                                format: int32
+                                type: integer
                               name:
                                 default: default
                                 type: string
@@ -7543,6 +7585,41 @@ spec:
                     type: object
                 type: object
               cat:
+                properties:
+                  resources:
+                    default:
+                      limits:
+                        cpu: 500m
+                        memory: 256Mi
+                      requests:
+                        cpu: 100m
+                        memory: 64Mi
+                    description: ResourceRequirements describes the compute resource
+                      requirements.
+                    properties:
+                      limits:
+                        additionalProperties:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
+                        description: 'Limits describes the maximum amount of compute
+                          resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                        type: object
+                      requests:
+                        additionalProperties:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
+                        description: 'Requests describes the minimum amount of compute
+                          resources required. If Requests is omitted for a container,
+                          it defaults to Limits if that is explicitly specified, otherwise
+                          to an implementation-defined value. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                        type: object
+                    type: object
                 type: object
               code:
                 properties:
@@ -8233,6 +8310,9 @@ spec:
                           items:
                             type: string
                           type: array
+                        maxMessageBytes:
+                          format: int32
+                          type: integer
                         name:
                           default: default
                           type: string
@@ -10060,6 +10140,9 @@ spec:
                           items:
                             type: string
                           type: array
+                        maxMessageBytes:
+                          format: int32
+                          type: integer
                         name:
                           default: default
                           type: string
diff --git a/config/quick-start.yaml b/config/quick-start.yaml
index 952a3acc..a27969f3 100644
--- a/config/quick-start.yaml
+++ b/config/quick-start.yaml
@@ -681,6 +681,42 @@ spec:
                           type: object
                       type: object
                     cat:
+                      properties:
+                        resources:
+                          default:
+                            limits:
+                              cpu: 500m
+                              memory: 256Mi
+                            requests:
+                              cpu: 100m
+                              memory: 64Mi
+                          description: ResourceRequirements describes the compute
+                            resource requirements.
+                          properties:
+                            limits:
+                              additionalProperties:
+                                anyOf:
+                                - type: integer
+                                - type: string
+                                pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                x-kubernetes-int-or-string: true
+                              description: 'Limits describes the maximum amount of
+                                compute resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                              type: object
+                            requests:
+                              additionalProperties:
+                                anyOf:
+                                - type: integer
+                                - type: string
+                                pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                x-kubernetes-int-or-string: true
+                              description: 'Requests describes the minimum amount
+                                of compute resources required. If Requests is omitted
+                                for a container, it defaults to Limits if that is
+                                explicitly specified, otherwise to an implementation-defined
+                                value. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                              type: object
+                          type: object
                       type: object
                     code:
                       properties:
@@ -1386,6 +1422,9 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              maxMessageBytes:
+                                format: int32
+                                type: integer
                               name:
                                 default: default
                                 type: string
@@ -3327,6 +3366,9 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              maxMessageBytes:
+                                format: int32
+                                type: integer
                               name:
                                 default: default
                                 type: string
@@ -7543,6 +7585,41 @@ spec:
                     type: object
                 type: object
               cat:
+                properties:
+                  resources:
+                    default:
+                      limits:
+                        cpu: 500m
+                        memory: 256Mi
+                      requests:
+                        cpu: 100m
+                        memory: 64Mi
+                    description: ResourceRequirements describes the compute resource
+                      requirements.
+                    properties:
+                      limits:
+                        additionalProperties:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
+                        description: 'Limits describes the maximum amount of compute
+                          resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                        type: object
+                      requests:
+                        additionalProperties:
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
+                        description: 'Requests describes the minimum amount of compute
+                          resources required. If Requests is omitted for a container,
+                          it defaults to Limits if that is explicitly specified, otherwise
+                          to an implementation-defined value. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
+                        type: object
+                    type: object
                 type: object
               code:
                 properties:
@@ -8233,6 +8310,9 @@ spec:
                           items:
                             type: string
                           type: array
+                        maxMessageBytes:
+                          format: int32
+                          type: integer
                         name:
                           default: default
                           type: string
@@ -10060,6 +10140,9 @@ spec:
                           items:
                             type: string
                           type: array
+                        maxMessageBytes:
+                          format: int32
+                          type: integer
                         name:
                           default: default
                           type: string
diff --git a/docs/BIG_PAYLOAD.md b/docs/BIG_PAYLOAD.md
new file mode 100644
index 00000000..63f8d8dd
--- /dev/null
+++ b/docs/BIG_PAYLOAD.md
@@ -0,0 +1,57 @@
+# Processing Big Payload >1MB
+Default Argo Dataflow configuration can process the upto 1MB payload. If you want to process bigger payload,
+you need to configure the resource for `step` and `sidecar`. We have tested below large payloads on HTTP, Kafka and Stan source.
+
+#### Resource Configuration:
+```yaml
+apiVersion: dataflow.argoproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: http
+spec: 
+  steps:
+  - cat:
+      resources:
+        requests:
+          memory: 1Gi
+    name: main
+    replicas: 5
+    restartPolicy: OnFailure
+    scale:
+      peekDelay: defaultPeekDelay
+      scalingDelay: defaultScalingDelay
+    serviceAccountName: pipeline
+    sidecar:
+      resources:
+        requests:
+          memory: 1Gi
+    sinks:
+    - log:
+        truncate: 32
+      name: log
+    sources:
+    - http: {}
+      name: default
+      retry:
+        cap: 0ms
+        duration: 100ms
+        factorPercentage: 200
+        jitterPercentage: 10
+        steps: 20
+```
+
+
+
+Results:
+   
+| Source | 1 Mb|5 Mb|10 Mb|
+|---|---|---|---|
+| HTTP |TPS 80 |TPS 35| TPS 17|
+| Kafka | TPS 74|TBS 19|TPS 11|
+|Stan|TPS 113|TPS 22|TPS 8|
+
+| Sink | 1 Mb|5 Mb|10 Mb|
+|---|---|---|---|
+| HTTP |TPS 81 |TPS 34|TPS 17|
+| Kafka | TPS 74|TPS 18|TPS 9|
+|Stan| TPS 69| TPS 6|TPS 3|    
diff --git a/runner/sidecar/sink/kafka/kafka.go b/runner/sidecar/sink/kafka/kafka.go
index 073ddfa5..8112dfb8 100644
--- a/runner/sidecar/sink/kafka/kafka.go
+++ b/runner/sidecar/sink/kafka/kafka.go
@@ -48,6 +48,11 @@ func New(ctx context.Context, secretInterface corev1.SecretInterface, x dfv1.Kaf
 		return nil, err
 	}
 
+	if x.MaxMessageBytes > 0 {
+		config.Producer.MaxMessageBytes = int(x.MaxMessageBytes)
+	}
+	logger.Info("Kafka config Producer.MaxMessageBytes", "value", config.Consumer.Fetch.Max)
+
 	producer, err := sarama.NewAsyncProducer(x.Brokers, config)
 	if err != nil {
 		return nil, err
diff --git a/runner/sidecar/source/kafka/kafka.go b/runner/sidecar/source/kafka/kafka.go
index 1e0bffee..40056fd4 100644
--- a/runner/sidecar/source/kafka/kafka.go
+++ b/runner/sidecar/source/kafka/kafka.go
@@ -33,6 +33,10 @@ func New(ctx context.Context, secretInterface corev1.SecretInterface, consumerGr
 	}
 	config.Consumer.MaxProcessingTime = 10 * time.Second
 	config.Consumer.Fetch.Max = 16 * config.Consumer.Fetch.Default
+	if x.Kafka.MaxMessageBytes > 0 {
+		config.Consumer.Fetch.Max = x.Kafka.MaxMessageBytes
+	}
+	logger.Info("Kafka config Consumer.Fetch.Max", "value", config.Consumer.Fetch.Max)
 	config.Consumer.Offsets.AutoCommit.Enable = false
 	if x.StartOffset == "First" {
 		config.Consumer.Offsets.Initial = sarama.OffsetOldest


Test patch:
diff --git a/api/v1alpha1/cat_test.go b/api/v1alpha1/cat_test.go
index 1e854d7f..906ef110 100644
--- a/api/v1alpha1/cat_test.go
+++ b/api/v1alpha1/cat_test.go
@@ -3,6 +3,9 @@ package v1alpha1
 import (
 	"testing"
 
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	"github.com/stretchr/testify/assert"
 )
 
@@ -10,4 +13,15 @@ func TestCat_getContainer(t *testing.T) {
 	x := Cat{}
 	c := x.getContainer(getContainerReq{})
 	assert.Equal(t, []string{"cat"}, c.Args)
+	assert.Equal(t, standardResources, c.Resources)
+
+	resource := v1.ResourceRequirements{
+		Requests: v1.ResourceList{
+
+			v1.ResourceMemory: resource.MustParse("2Gi"),
+		},
+	}
+	x = Cat{AbstractStep: AbstractStep{Resources: resource}}
+	c = x.getContainer(getContainerReq{})
+	assert.Equal(t, resource.Requests.Memory(), c.Resources.Requests.Memory())
 }
diff --git a/api/v1alpha1/step_types_test.go b/api/v1alpha1/step_types_test.go
index db521951..3d8639cb 100644
--- a/api/v1alpha1/step_types_test.go
+++ b/api/v1alpha1/step_types_test.go
@@ -22,7 +22,7 @@ func TestStep_GetPodSpec(t *testing.T) {
 				{Name: "ARGO_DATAFLOW_NAMESPACE", Value: "my-ns"},
 				{Name: "ARGO_DATAFLOW_PIPELINE_NAME", Value: "my-pl"},
 				{Name: "ARGO_DATAFLOW_REPLICA", Value: fmt.Sprintf("%d", replica)},
-				{Name: "ARGO_DATAFLOW_STEP", Value: `{"metadata":{"creationTimestamp":null},"spec":{"name":"main","cat":{},"scale":{},"sidecar":{"resources":{}}},"status":{"phase":"","replicas":0,"lastScaledAt":null}}`},
+				{Name: "ARGO_DATAFLOW_STEP", Value: `{"metadata":{"creationTimestamp":null},"spec":{"name":"main","cat":{"resources":{"limits":{"cpu":"500m","memory":"256Mi"},"requests":{"cpu":"100m","memory":"64Mi"}}},"scale":{},"sidecar":{"resources":{}}},"status":{"phase":"","replicas":0,"lastScaledAt":null}}`},
 				{Name: "ARGO_DATAFLOW_UPDATE_INTERVAL", Value: "1m0s"},
 				{Name: "GODEBUG"},
 			}
@@ -44,7 +44,7 @@ func TestStep_GetPodSpec(t *testing.T) {
 					Step{
 						Spec: StepSpec{
 							Name: "main",
-							Cat:  &Cat{},
+							Cat:  &Cat{AbstractStep{Resources: standardResources}},
 						},
 					},
 					GetPodSpecReq{
diff --git a/config/apps/testapi/test.yaml b/config/apps/testapi/test.yaml
new file mode 100644
index 00000000..be88ec63
--- /dev/null
+++ b/config/apps/testapi/test.yaml
@@ -0,0 +1,37 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: testapi
+  namespace: argo-dataflow-system
+spec:
+  ports:
+  - port: 80
+    targetPort: 8378
+  selector:
+    app: testapi
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: testapi
+  namespace: argo-dataflow-system
+spec:
+  selector:
+    matchLabels:
+      app: testapi
+  serviceName: testapi
+  template:
+    metadata:
+      labels:
+        app: testapi
+    spec:
+      containers:
+      - image: quay.io/argoproj/dataflow-testapi
+        imagePullPolicy: IfNotPresent
+        name: main
+        ports:
+        - containerPort: 8378
+        readinessProbe:
+          httpGet:
+            path: /ready
+            port: 8378
diff --git a/test/http-stress/http_stress_test.go b/test/http-stress/http_stress_test.go
index 71a41dc7..7f6e5c88 100644
--- a/test/http-stress/http_stress_test.go
+++ b/test/http-stress/http_stress_test.go
@@ -5,10 +5,12 @@ package http_stress
 import (
 	"testing"
 
-	. "github.com/argoproj-labs/argo-dataflow/test/stress"
-
 	. "github.com/argoproj-labs/argo-dataflow/api/v1alpha1"
 	. "github.com/argoproj-labs/argo-dataflow/test"
+	. "github.com/argoproj-labs/argo-dataflow/test/stress"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -19,11 +21,22 @@ func TestHTTPSourceStress(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{Name: "http"},
 		Spec: PipelineSpec{
 			Steps: []StepSpec{{
-				Name:     "main",
-				Cat:      &Cat{},
+				Name: "main",
+				Cat: &Cat{
+					AbstractStep: AbstractStep{Resources: v1.ResourceRequirements{
+						Requests: v1.ResourceList{
+							v1.ResourceMemory: resource.MustParse("1Gi"),
+						},
+					}},
+				},
 				Replicas: Params.Replicas,
 				Sources:  []Source{{HTTP: &HTTPSource{}}},
 				Sinks:    []Sink{DefaultLogSink},
+				Sidecar: Sidecar{Resources: v1.ResourceRequirements{
+					Requests: v1.ResourceList{
+						v1.ResourceMemory: resource.MustParse("1Gi"),
+					},
+				}},
 			}},
 		},
 	})
@@ -51,11 +64,20 @@ func TestHTTPSinkStress(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{Name: "http"},
 		Spec: PipelineSpec{
 			Steps: []StepSpec{{
-				Name:     "main",
-				Cat:      &Cat{},
+				Name: "main",
+				Cat: &Cat{AbstractStep: AbstractStep{Resources: v1.ResourceRequirements{
+					Requests: v1.ResourceList{
+						v1.ResourceMemory: resource.MustParse("1Gi"),
+					},
+				}}},
 				Replicas: Params.Replicas,
 				Sources:  []Source{{HTTP: &HTTPSource{}}},
 				Sinks:    []Sink{{HTTP: &HTTPSink{URL: "http://testapi/count/incr"}}, DefaultLogSink},
+				Sidecar: Sidecar{Resources: v1.ResourceRequirements{
+					Requests: v1.ResourceList{
+						v1.ResourceMemory: resource.MustParse("1Gi"),
+					},
+				}},
 			}},
 		},
 	})
diff --git a/test/kafka-stress/kafka_stress_test.go b/test/kafka-stress/kafka_stress_test.go
index aac077c6..b81af67e 100644
--- a/test/kafka-stress/kafka_stress_test.go
+++ b/test/kafka-stress/kafka_stress_test.go
@@ -5,6 +5,9 @@ package kafka_stress
 import (
 	"testing"
 
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	. "github.com/argoproj-labs/argo-dataflow/test/stress"
 
 	. "github.com/argoproj-labs/argo-dataflow/api/v1alpha1"
@@ -18,15 +21,21 @@ func TestKafkaSourceStress(t *testing.T) {
 	defer Setup(t)()
 
 	topic := CreateKafkaTopic()
-
+	msgSize := int32(50000000)
 	CreatePipeline(Pipeline{
 		ObjectMeta: metav1.ObjectMeta{Name: "kafka"},
 		Spec: PipelineSpec{
 			Steps: []StepSpec{{
-				Name:     "main",
-				Cat:      &Cat{},
+				Name: "main",
+				Cat: &Cat{
+					AbstractStep: AbstractStep{Resources: v1.ResourceRequirements{
+						Requests: v1.ResourceList{
+							v1.ResourceMemory: resource.MustParse("1Gi"),
+						},
+					}},
+				},
 				Replicas: Params.Replicas,
-				Sources:  []Source{{Kafka: &KafkaSource{Kafka: Kafka{Topic: topic}}}},
+				Sources:  []Source{{Kafka: &KafkaSource{Kafka: Kafka{Topic: topic, KafkaConfig: KafkaConfig{MaxMessageBytes: msgSize}}}}},
 				Sinks:    []Sink{DefaultLogSink},
 			}},
 		},
@@ -52,17 +61,23 @@ func TestKafkaSinkStress(t *testing.T) {
 
 	topic := CreateKafkaTopic()
 	sinkTopic := CreateKafkaTopic()
-
+	msgSize := int32(50000000)
 	CreatePipeline(Pipeline{
 		ObjectMeta: metav1.ObjectMeta{Name: "kafka"},
 		Spec: PipelineSpec{
 			Steps: []StepSpec{{
-				Name:     "main",
-				Cat:      &Cat{},
+				Name: "main",
+				Cat: &Cat{
+					AbstractStep: AbstractStep{Resources: v1.ResourceRequirements{
+						Requests: v1.ResourceList{
+							v1.ResourceMemory: resource.MustParse("1Gi"),
+						},
+					}},
+				},
 				Replicas: Params.Replicas,
-				Sources:  []Source{{Kafka: &KafkaSource{Kafka: Kafka{Topic: topic}}}},
+				Sources:  []Source{{Kafka: &KafkaSource{Kafka: Kafka{Topic: topic, KafkaConfig: KafkaConfig{MaxMessageBytes: msgSize}}}}},
 				Sinks: []Sink{
-					{Kafka: &KafkaSink{Async: Params.Async, Kafka: Kafka{Topic: sinkTopic}}},
+					{Kafka: &KafkaSink{Async: Params.Async, Kafka: Kafka{Topic: sinkTopic, KafkaConfig: KafkaConfig{MaxMessageBytes: msgSize}}}},
 					DefaultLogSink,
 				},
 			}},
diff --git a/test/stan-stress/stan_stress_test.go b/test/stan-stress/stan_stress_test.go
index 9807b968..e9dae0be 100644
--- a/test/stan-stress/stan_stress_test.go
+++ b/test/stan-stress/stan_stress_test.go
@@ -5,6 +5,9 @@ package stan_stress
 import (
 	"testing"
 
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	. "github.com/argoproj-labs/argo-dataflow/test/stress"
 
 	. "github.com/argoproj-labs/argo-dataflow/api/v1alpha1"
@@ -28,6 +31,11 @@ func TestStanSourceStress(t *testing.T) {
 				Replicas: Params.Replicas,
 				Sources:  []Source{{STAN: &STAN{Subject: subject}}},
 				Sinks:    []Sink{DefaultLogSink},
+				Sidecar: Sidecar{Resources: v1.ResourceRequirements{
+					Requests: v1.ResourceList{
+						v1.ResourceMemory: resource.MustParse("1Gi"),
+					},
+				}},
 			}},
 		},
 	})
@@ -62,6 +70,11 @@ func TestStanSinkStress(t *testing.T) {
 				Replicas: Params.Replicas,
 				Sources:  []Source{{STAN: &STAN{Subject: subject}}},
 				Sinks:    []Sink{{STAN: &STAN{Subject: sinkSubject}}, DefaultLogSink},
+				Sidecar: Sidecar{Resources: v1.ResourceRequirements{
+					Requests: v1.ResourceList{
+						v1.ResourceMemory: resource.MustParse("1Gi"),
+					},
+				}},
 			}},
 		},
 	})
@@ -72,7 +85,7 @@ func TestStanSinkStress(t *testing.T) {
 
 	WaitForPod()
 
-	n := 10000
+	n := Params.N
 	prefix := "stan-sink-stress"
 	defer StartTPSReporter(t, "main", prefix, n)()
 
diff --git a/test/stan-stress/test-results.json b/test/stan-stress/test-results.json
index a015a754..dc6fab2b 100644
--- a/test/stan-stress/test-results.json
+++ b/test/stan-stress/test-results.json
@@ -1,6 +1,10 @@
 {
   "TestStanSinkStress/.tps": 200,
   "TestStanSinkStress/N=10,messageSize=100.tps": 150,
+  "TestStanSinkStress/currentContext=gke_jesse-sb_us-central1-c_dataflow-test,N=10,messageSize=5000000.tps": 200,
+  "TestStanSinkStress/currentContext=gke_jesse-sb_us-central1-c_dataflow-test,N=100,messageSize=10.tps": 350,
+  "TestStanSinkStress/currentContext=gke_jesse-sb_us-central1-c_dataflow-test,N=100,messageSize=1000000.tps": 50,
+  "TestStanSinkStress/currentContext=gke_jesse-sb_us-central1-c_dataflow-test,N=100,messageSize=10000000.tps": 0,
   "TestStanSinkStress/replicas=2.tps": 300,
   "TestStanSinkStress/replicas=2tps": 250,
   "TestStanSinkStress/replicas=3.tps": 300,
@@ -9,6 +13,11 @@
   "TestStanSourceStress/.tps": 1050,
   "TestStanSourceStress/N=10,messageSize=100.tps": 850,
   "TestStanSourceStress/N=10,messageSize=1000.tps": 400,
+  "TestStanSourceStress/currentContext=gke_jesse-sb_us-central1-c_dataflow-test,N=10,messageSize=5000000.tps": 40300,
+  "TestStanSourceStress/currentContext=gke_jesse-sb_us-central1-c_dataflow-test,N=100,messageSize=10.tps": 2000,
+  "TestStanSourceStress/currentContext=gke_jesse-sb_us-central1-c_dataflow-test,N=100,messageSize=1000000.tps": 100,
+  "TestStanSourceStress/currentContext=gke_jesse-sb_us-central1-c_dataflow-test,N=100,messageSize=10000000.tps": 0,
+  "TestStanSourceStress/currentContext=gke_jesse-sb_us-central1-c_dataflow-test,N=100,messageSize=5000000.tps": 0,
   "TestStanSourceStress/messageSize=1000.tps": 950,
   "TestStanSourceStress/replicas=2.tps": 1300,
   "TestStanSourceStress/replicas=3.tps": 750,
diff --git a/testapi/kafka.go b/testapi/kafka.go
index 84901a15..4de8956b 100644
--- a/testapi/kafka.go
+++ b/testapi/kafka.go
@@ -14,6 +14,7 @@ import (
 func init() {
 	sarama.Logger = log.New(os.Stdout, "", log.LstdFlags)
 	config := sarama.NewConfig()
+	config.Producer.MaxMessageBytes = 50000000
 	config.ClientID = "dataflow-testapi"
 	addrs := []string{"kafka-broker:9092"}