ha.diff

diff --git a/control-plane/cmd/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
index 5447ae63..13b8c84c 100644
--- a/control-plane/cmd/kafka-controller/main.go
+++ b/control-plane/cmd/kafka-controller/main.go
@@ -29,10 +29,12 @@ import (
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
+	channelv2 "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel/v2"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumer"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/sink"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/source"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/trigger"
+	sourcev2 "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/source/v2"
+	triggerv2 "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/trigger/v2"
 )
 
 const (
@@ -58,7 +60,12 @@ func main() {
 
 	sourceEnv, err := config.GetEnvConfig("SOURCE")
 	if err != nil {
-		log.Fatal("cannot process environment variables with prefix SINK", err)
+		log.Fatal("cannot process environment variables with prefix SOURCE", err)
+	}
+
+	consumerEnv, err := config.GetEnvConfig("SOURCE")
+	if err != nil {
+		log.Fatal("cannot process environment variables with prefix CONSUMER", err)
 	}
 
 	sharedmain.MainNamed(signals.NewContext(), component,
@@ -75,7 +82,7 @@ func main() {
 		injection.NamedControllerConstructor{
 			Name: "trigger-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return trigger.NewController(ctx, watcher, brokerEnv)
+				return triggerv2.NewController(ctx, brokerEnv)
 			},
 		},
 
@@ -83,7 +90,7 @@ func main() {
 		injection.NamedControllerConstructor{
 			Name: "channel-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return channel.NewController(ctx, watcher, channelEnv)
+				return channelv2.NewController(ctx, channelEnv)
 			},
 		},
 
@@ -99,7 +106,23 @@ func main() {
 		injection.NamedControllerConstructor{
 			Name: "source-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return source.NewController(ctx, watcher, sourceEnv)
+				return sourcev2.NewController(ctx, sourceEnv)
+			},
+		},
+
+		// ConsumerGroup controller
+		injection.NamedControllerConstructor{
+			Name: "consumergroup-controller",
+			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
+				return consumergroup.NewController(ctx)
+			},
+		},
+
+		// Consumer controller
+		injection.NamedControllerConstructor{
+			Name: "consumer-controller",
+			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
+				return consumer.NewController(ctx, consumerEnv)
 			},
 		},
 	)
diff --git a/data-plane/config/source/500-dispatcher.yaml b/data-plane/config/source/500-dispatcher.yaml
index 3be219bf..9866ab86 100644
--- a/data-plane/config/source/500-dispatcher.yaml
+++ b/data-plane/config/source/500-dispatcher.yaml
@@ -15,7 +15,7 @@
 # limitations under the License.
 
 apiVersion: apps/v1
-kind: Deployment
+kind: StatefulSet
 metadata:
   name: kafka-source-dispatcher
   namespace: knative-eventing
@@ -23,6 +23,8 @@ metadata:
     app: kafka-source-dispatcher
     kafka.eventing.knative.dev/release: devel
 spec:
+  serviceName: kafka-source-dispatcher
+  podManagementPolicy: "Parallel"
   selector:
     matchLabels:
       app: kafka-source-dispatcher
@@ -31,6 +33,7 @@ spec:
       name: kafka-source-dispatcher
       labels:
         app: kafka-source-dispatcher
+        app.kubernetes.io/component: kafka-dispatcher
         kafka.eventing.knative.dev/release: devel
     spec:
       serviceAccountName: knative-kafka-source-data-plane
@@ -44,8 +47,8 @@ spec:
             - mountPath: /etc/config
               name: config-kafka-source-data-plane
               readOnly: true
-            - mountPath: /etc/sources
-              name: kafka-source-sources
+            - mountPath: /etc/contract-resources
+              name: contract-resources
               readOnly: true
             - mountPath: /tmp
               name: cache
@@ -73,7 +76,7 @@ spec:
             - name: WEBCLIENT_CONFIG_FILE_PATH
               value: /etc/config/config-kafka-source-webclient.properties
             - name: DATA_PLANE_CONFIG_FILE_PATH
-              value: /etc/sources/data
+              value: /etc/contract-resources/data
             - name: EGRESSES_INITIAL_CAPACITY
               value: "20"
             - name: INSTANCE_ID
@@ -102,7 +105,16 @@ spec:
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
-          # TODO set resources (limits and requests)
+
+          resources:
+            requests:
+              cpu: 1000m
+              # 600Mi for virtual replicas + 100Mi overhead
+              memory: 700Mi
+            limits:
+              cpu: 2000m
+              memory: 1000Mi
+
           livenessProbe:
             failureThreshold: 3
             httpGet:
@@ -133,9 +145,6 @@ spec:
         - name: config-kafka-source-data-plane
           configMap:
             name: config-kafka-source-data-plane
-        - name: kafka-source-sources
-          configMap:
-            name: kafka-source-sources
         - name: cache
           emptyDir: { }
         - name: kafka-config-logging
diff --git a/hack/data-plane.sh b/hack/data-plane.sh
index c8ac4a06..f6de716d 100755
--- a/hack/data-plane.sh
+++ b/hack/data-plane.sh
@@ -152,8 +152,8 @@ function data_plane_build_push() {
 function replace_images() {
   local file=$1
 
-  sed -i "s|\${KNATIVE_KAFKA_DISPATCHER_IMAGE}|${KNATIVE_KAFKA_DISPATCHER_IMAGE}|g" "${file}" &&
-    sed -i "s|\${KNATIVE_KAFKA_RECEIVER_IMAGE}|${KNATIVE_KAFKA_RECEIVER_IMAGE}|g" "${file}"
+  /usr/bin/sed -i '' "s|\${KNATIVE_KAFKA_DISPATCHER_IMAGE}|${KNATIVE_KAFKA_DISPATCHER_IMAGE}|g" "${file}" &&
+    /usr/bin/sed -i '' "s|\${KNATIVE_KAFKA_RECEIVER_IMAGE}|${KNATIVE_KAFKA_RECEIVER_IMAGE}|g" "${file}"
 
   return $?
 }
@@ -164,6 +164,7 @@ function k8s() {
   echo "Dispatcher image ---> ${KNATIVE_KAFKA_DISPATCHER_IMAGE}"
   echo "Receiver image   ---> ${KNATIVE_KAFKA_RECEIVER_IMAGE}"
 
+
   ko resolve ${KO_FLAGS} -Rf ${SOURCE_DATA_PLANE_CONFIG_DIR} | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_SOURCE_ARTIFACT}"
   ko resolve ${KO_FLAGS} -Rf ${BROKER_DATA_PLANE_CONFIG_DIR} | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_BROKER_ARTIFACT}"
   ko resolve ${KO_FLAGS} -Rf ${SINK_DATA_PLANE_CONFIG_DIR} | "${LABEL_YAML_CMD[@]}" >>"${EVENTING_KAFKA_SINK_ARTIFACT}"
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index f883afbc..93a3b841 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -158,12 +158,12 @@ function test_setup() {
 
   setup_kafka_channel_auth || fail_test "Failed to apply channel auth configuration ${EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO}"
 
-  kubectl rollout restart deployment -n knative-eventing kafka-source-dispatcher
+  kubectl rollout restart statefulset -n knative-eventing kafka-source-dispatcher
   kubectl rollout restart deployment -n knative-eventing kafka-broker-receiver
-  kubectl rollout restart deployment -n knative-eventing kafka-broker-dispatcher
+  kubectl rollout restart statefulset -n knative-eventing kafka-broker-dispatcher
   kubectl rollout restart deployment -n knative-eventing kafka-sink-receiver
   kubectl rollout restart deployment -n knative-eventing kafka-channel-receiver
-  kubectl rollout restart deployment -n knative-eventing kafka-channel-dispatcher
+  kubectl rollout restart statefulset -n knative-eventing kafka-channel-dispatcher
 }
 
 function test_teardown() {
@@ -215,7 +215,7 @@ function delete_sacura() {
 
 function export_logs_continuously() {
 
-  labels=("kafka-broker-dispatcher" "kafka-broker-receiver" "kafka-sink-receiver" "kafka-channel-receiver" "kafka-channel-dispatcher" "kafka-source-dispatcher" "kafka-webhook-eventing" "kafka-controller")
+  labels=("kafka-broker-dispatcher" "kafka-broker-receiver" "kafka-sink-receiver" "kafka-channel-receiver" "kafka-channel-dispatcher" "kafka-webhook-eventing" "kafka-controller")
 
   mkdir -p "$ARTIFACTS/${SYSTEM_NAMESPACE}"
 
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/state/state.go b/vendor/knative.dev/eventing/pkg/scheduler/state/state.go
index f3b7f12f..4e98d4c1 100644
--- a/vendor/knative.dev/eventing/pkg/scheduler/state/state.go
+++ b/vendor/knative.dev/eventing/pkg/scheduler/state/state.go
@@ -273,7 +273,10 @@ func (s *stateBuilder) State(reserved map[types.NamespacedName]map[string]int32)
 			// Account for reserved vreplicas
 			vreplicas = withReserved(vpod.GetKey(), podName, vreplicas, reserved)
 
-			free, last = s.updateFreeCapacity(free, last, podName, vreplicas)
+			free, last, err = s.updateFreeCapacity(free, last, podName, vreplicas)
+			if err != nil {
+				return nil, err
+			}
 
 			withPlacement[vpod.GetKey()][podName] = true
 
@@ -317,7 +320,10 @@ func (s *stateBuilder) State(reserved map[types.NamespacedName]map[string]int32)
 				}
 			}
 
-			free, last = s.updateFreeCapacity(free, last, podName, rvreplicas)
+			free, last, err = s.updateFreeCapacity(free, last, podName, rvreplicas)
+			if err != nil {
+				return nil, err
+			}
 		}
 	}
 
@@ -327,23 +333,23 @@ func (s *stateBuilder) State(reserved map[types.NamespacedName]map[string]int32)
 		PodSpread: podSpread, NodeSpread: nodeSpread, ZoneSpread: zoneSpread}, nil
 }
 
-func (s *stateBuilder) updateFreeCapacity(free []int32, last int32, podName string, vreplicas int32) ([]int32, int32) {
+func (s *stateBuilder) updateFreeCapacity(free []int32, last int32, podName string, vreplicas int32) ([]int32, int32, error) {
 	ordinal := OrdinalFromPodName(podName)
 	free = grow(free, ordinal, s.capacity)
 
 	free[ordinal] -= vreplicas
 
-	// Assert the pod is not overcommitted
 	if free[ordinal] < 0 {
-		// This should not happen anymore. Log as an error but do not interrupt the current scheduling.
+		// Pod is overcommitted. Return error so that vpod is rescheduled
 		s.logger.Errorw("pod is overcommitted", zap.String("podName", podName), zap.Int32("free", free[ordinal]))
+		return free, last, fmt.Errorf("pod is overcommitted - need to resechedule")
 	}
 
 	if ordinal > last && free[ordinal] != s.capacity {
 		last = ordinal
 	}
 
-	return free, last
+	return free, last, nil
 }
 
 func grow(slice []int32, ordinal int32, def int32) []int32 {
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/statefulset/autoscaler.go b/vendor/knative.dev/eventing/pkg/scheduler/statefulset/autoscaler.go
index 6b66a058..27bf72e8 100644
--- a/vendor/knative.dev/eventing/pkg/scheduler/statefulset/autoscaler.go
+++ b/vendor/knative.dev/eventing/pkg/scheduler/statefulset/autoscaler.go
@@ -19,6 +19,7 @@ package statefulset
 import (
 	"context"
 	"math"
+	"strings"
 	"sync"
 	"time"
 
@@ -118,7 +119,7 @@ func (a *autoscaler) syncAutoscale(ctx context.Context, attemptScaleDown bool, p
 
 func (a *autoscaler) doautoscale(ctx context.Context, attemptScaleDown bool, pending int32) error {
 	state, err := a.stateAccessor.State(nil)
-	if err != nil {
+	if err != nil && !strings.Contains(err.Error(), "pod overcommitted") {
 		a.logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
 		return err
 	}
diff --git a/vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go b/vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go
index 643bae00..acb63c82 100644
--- a/vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go
+++ b/vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go
@@ -166,18 +166,19 @@ func (s *StatefulSetScheduler) scheduleVPod(vpod scheduler.VPod) ([]duckv1alpha1
 	logger := s.logger.With("key", vpod.GetKey())
 	logger.Info("scheduling")
 
+	placements := vpod.GetPlacements()
+	existingPlacements := placements
+	var left int32
+
 	// Get the current placements state
 	// Quite an expensive operation but safe and simple.
 	state, err := s.stateAccessor.State(s.reserved)
 	if err != nil {
 		logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
+		s.makeZeroPlacements(vpod, placements)
 		return nil, err
 	}
 
-	placements := vpod.GetPlacements()
-	existingPlacements := placements
-	var left int32
-
 	// The scheduler when policy type is
 	// Policy: MAXFILLUP (SchedulerPolicyType == MAXFILLUP)
 	// - allocates as many vreplicas as possible to the same pod(s)
@@ -273,6 +274,7 @@ func (s *StatefulSetScheduler) removeReplicasWithPolicy(vpod scheduler.VPod, dif
 		state, err := s.stateAccessor.State(s.reserved)
 		if err != nil {
 			logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
+			s.makeZeroPlacements(vpod, placements)
 			return placements
 		}
 
@@ -342,6 +344,7 @@ func (s *StatefulSetScheduler) addReplicasWithPolicy(vpod scheduler.VPod, diff i
 		state, err := s.stateAccessor.State(s.reserved)
 		if err != nil {
 			logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
+			s.makeZeroPlacements(vpod, placements)
 			return placements, diff
 		}