fix merge conflicts

Signed-off-by: Calum Murray <cmurray@redhat.com>

.github/workflows/knative-go-test.yaml

@@ -15,3 +15,5 @@ on:
 jobs:
   test:
     uses: knative/actions/.github/workflows/reusable-go-test.yaml@main
+    secrets:
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}

OWNERS_ALIASES

@@ -166,7 +166,6 @@ aliases:
   - knative-prow-robot
   - knative-prow-updater-robot
   - knative-test-reporter-robot
-  - krsna-m
   - nainaz
   - psschwei
   - salaboy
@@ -202,7 +201,6 @@ aliases:
   - upodroid
   productivity-writers:
   - cardil
-  - krsna-m
   - upodroid
   security-guard-approvers:
   - davidhadas
@@ -246,7 +244,6 @@ aliases:
   - davidhadas
   - dprotaso
   - dsimansk
-  - krsna-m
   - psschwei
   ux-wg-leads:
   - cali0707

control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/register.go

@@ -75,3 +75,15 @@ func IsKnownStatefulSet(name string) bool {
 		name == ChannelStatefulSetName ||
 		name == BrokerStatefulSetName
 }
+
+func GetOwnerKindFromStatefulSetName(name string) (string, bool) {
+	switch name {
+	case SourceStatefulSetName:
+		return "KafkaSource", true
+	case ChannelStatefulSetName:
+		return "KafkaChannel", true
+	case BrokerStatefulSetName:
+		return "Trigger", true
+	}
+	return "", false
+}

control-plane/pkg/contract/extensions.go

@@ -16,7 +16,16 @@
 
 package contract
 
+import (
+	"github.com/google/go-cmp/cmp"
+	"google.golang.org/protobuf/testing/protocmp"
+)
+
 // IncrementGeneration increments Generation.
 func (x *Contract) IncrementGeneration() {
 	x.Generation++
 }
+
+func SemanticEqual(ct1 *Contract, ct2 *Contract) bool {
+	return cmp.Equal(ct1, ct2, protocmp.Transform(), protocmp.IgnoreFields(ct1, "generation"))
+}

control-plane/pkg/core/config/egress.go

@@ -47,24 +47,17 @@ const (
 )
 
 // AddOrUpdateEgressConfig adds or updates the given egress to the given contract at the specified indexes.
-func AddOrUpdateEgressConfig(ct *contract.Contract, resourceIndex int, egress *contract.Egress, egressIndex int) int {
+func AddOrUpdateEgressConfig(ct *contract.Contract, resourceIndex int, egress *contract.Egress, egressIndex int) {
 
 	if egressIndex != NoEgress {
-		prev := ct.Resources[resourceIndex].Egresses[egressIndex]
 		ct.Resources[resourceIndex].Egresses[egressIndex] = egress
-
-		if proto.Equal(prev, egress) {
-			return EgressUnchanged
-		}
-		return EgressChanged
+		return
 	}
 
 	ct.Resources[resourceIndex].Egresses = append(
 		ct.Resources[resourceIndex].Egresses,
 		egress,
 	)
-
-	return EgressChanged
 }
 
 // AddOrUpdateEgressConfigForResource adds or updates the given egress to the given contract at the specified indexes.

control-plane/pkg/core/config/egress_test.go

@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/testing/protocmp"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
@@ -210,13 +211,17 @@ func TestAddOrUpdateEgressConfig(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := AddOrUpdateEgressConfig(tt.givenCt, tt.brokerIndex, tt.egress, tt.egressIndex); got != tt.changed {
-				t.Errorf("AddOrUpdateEgressConfig() = %v, want %v", got, tt.changed)
-			}
+			before := proto.Clone(tt.givenCt).(*contract.Contract)
+			AddOrUpdateEgressConfig(tt.givenCt, tt.brokerIndex, tt.egress, tt.egressIndex)
 
 			if diff := cmp.Diff(tt.wantCt, tt.givenCt, protocmp.Transform()); diff != "" {
 				t.Errorf("(-want, +got) %s", diff)
 			}
+
+			gotEqual := contract.SemanticEqual(before, tt.wantCt)
+			if expectedEqual := contract.SemanticEqual(before, tt.wantCt); expectedEqual != gotEqual {
+				t.Errorf("expectEqual want %v got %v", expectedEqual, gotEqual)
+			}
 		})
 	}
 }

control-plane/pkg/core/config/resource.go

@@ -18,7 +18,6 @@ package config
 
 import (
 	"go.uber.org/zap"
-	"google.golang.org/protobuf/proto"
 	"k8s.io/apimachinery/pkg/types"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
@@ -41,37 +40,22 @@ func FindResource(contract *contract.Contract, resource types.UID) int {
 	return resourceIndex
 }
 
-const (
-	ResourceChanged = iota
-	ResourceUnchanged
-)
-
 func SetResourceEgressesFromContract(contract *contract.Contract, resource *contract.Resource, index int) {
 	if index != NoResource {
 		resource.Egresses = contract.Resources[index].Egresses
 	}
 }
 
 // AddOrUpdateResourceConfig adds or updates the given resourceConfig to the given resources at the specified index.
-func AddOrUpdateResourceConfig(contract *contract.Contract, resource *contract.Resource, index int, logger *zap.Logger) int {
-
+func AddOrUpdateResourceConfig(contract *contract.Contract, resource *contract.Resource, index int, logger *zap.Logger) {
 	if index != NoResource {
 		logger.Debug("Resource exists", zap.Int("index", index))
-
-		prev := contract.Resources[index]
 		contract.Resources[index] = resource
-
-		if proto.Equal(prev, resource) {
-			return ResourceUnchanged
-		}
-		return ResourceChanged
+		return
 	}
 
 	logger.Debug("Resource doesn't exist")
-
 	contract.Resources = append(contract.Resources, resource)
-
-	return ResourceChanged
 }
 
 // DeleteResource deletes the resource at the given index from Resources.

control-plane/pkg/core/config/resource_test.go

@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/testing/protocmp"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
@@ -86,7 +87,6 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 		newResource  *contract.Resource
 		index        int
 		wantContract *contract.Contract
-		changed      int
 	}{
 		{
 			name: "resource not found - add resource",
@@ -314,8 +314,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 					ContentMode: contract.ContentMode_STRUCTURED,
 				},
 			},
-			index:   0,
-			changed: ResourceUnchanged,
+			index: 0,
 			wantContract: &contract.Contract{
 				Resources: []*contract.Resource{
 					{
@@ -346,13 +345,16 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			changed := AddOrUpdateResourceConfig(tt.haveContract, tt.newResource, tt.index, zap.NewNop())
+			before := proto.Clone(tt.haveContract).(*contract.Contract)
+			AddOrUpdateResourceConfig(tt.haveContract, tt.newResource, tt.index, zap.NewNop())
 
 			if diff := cmp.Diff(tt.wantContract, tt.haveContract, protocmp.Transform()); diff != "" {
 				t.Errorf("(-want, +got) %s", diff)
 			}
-			if changed != tt.changed {
-				t.Errorf("Changed want %d got %d", tt.changed, changed)
+
+			gotEqual := contract.SemanticEqual(before, tt.wantContract)
+			if expectedEqual := contract.SemanticEqual(before, tt.wantContract); expectedEqual != gotEqual {
+				t.Errorf("expectEqual want %v got %v", expectedEqual, gotEqual)
 			}
 		})
 	}

control-plane/pkg/reconciler/base/reconciler.go

@@ -33,6 +33,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/utils/pointer"
+	"knative.dev/pkg/logging"
 	"knative.dev/pkg/tracker"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
@@ -235,7 +236,21 @@ func GetDataPlaneConfigMapData(logger *zap.Logger, dataPlaneConfigMap *corev1.Co
 	return ct, nil
 }
 
+func CompareSemanticEqual(ctx context.Context, ct *contract.Contract, existing *corev1.ConfigMap, format string) bool {
+	existingCt, err := GetDataPlaneConfigMapData(logging.FromContext(ctx).Desugar(), existing, format)
+	if existingCt != nil && err == nil {
+		return contract.SemanticEqual(existingCt, ct)
+	}
+	return false
+}
+
 func (r *Reconciler) UpdateDataPlaneConfigMap(ctx context.Context, contract *contract.Contract, configMap *corev1.ConfigMap) error {
+	if CompareSemanticEqual(ctx, contract, configMap, r.ContractConfigMapFormat) {
+		return nil
+	}
+
+	// Resource changed, increment contract generation.
+	coreconfig.IncrementContractGeneration(contract)
 
 	var data []byte
 	var err error
@@ -378,9 +393,6 @@ func (r *Reconciler) DeleteResource(ctx context.Context, logger *zap.Logger, uui
 
 		logger.Debug("Resource deleted", zap.Int("index", resourceIndex))
 
-		// Resource changed, increment contract generation.
-		coreconfig.IncrementContractGeneration(ct)
-
 		// Update the configuration map with the new contract data.
 		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
 			return err