Repository: knative-sandbox/eventing-kafka (go)

Issue:
Make v1beta1 storage version for KafkaChannel
Make v1beta1 storage version for KafkaChannel

Example WIP PR: https://github.com/knative/eventing-contrib/pull/1504

Part of umbrella issue, #70 

Pull request description:
Upgraded storage version of kafkachannel
Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

Fixes #73

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- 🧽 Upgraded storage version of kafka channel

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
🧽 Update or clean up current behavior
```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


Patch:
diff --git a/config/channel/resources/kafkachannel-crd.yaml b/config/channel/resources/kafkachannel-crd.yaml
index 78a0cbf73..fcf1ae773 100644
--- a/config/channel/resources/kafkachannel-crd.yaml
+++ b/config/channel/resources/kafkachannel-crd.yaml
@@ -27,7 +27,7 @@ spec:
     - &version
       name: v1alpha1
       served: true
-      storage: true
+      storage: false
       subresources:
         status: { }
       schema:
@@ -54,7 +54,7 @@ spec:
     - <<: *version
       name: v1beta1
       served: true
-      storage: false
+      storage: true
       schema:
         openAPIV3Schema:
           <<: *openAPIV3Schema
diff --git a/pkg/apis/messaging/v1alpha1/kafka_channel_defaults.go b/pkg/apis/messaging/v1alpha1/kafka_channel_defaults.go
index 6180798a1..ee951b551 100644
--- a/pkg/apis/messaging/v1alpha1/kafka_channel_defaults.go
+++ b/pkg/apis/messaging/v1alpha1/kafka_channel_defaults.go
@@ -33,7 +33,7 @@ func (c *KafkaChannel) SetDefaults(ctx context.Context) {
 		c.Annotations = make(map[string]string)
 	}
 	if _, ok := c.Annotations[messaging.SubscribableDuckVersionAnnotation]; !ok {
-		c.Annotations[messaging.SubscribableDuckVersionAnnotation] = "v1alpha1"
+		c.Annotations[messaging.SubscribableDuckVersionAnnotation] = "v1beta1"
 	}
 
 	c.Spec.SetDefaults(ctx)
diff --git a/pkg/apis/sources/v1beta1/kafka_defaults.go b/pkg/apis/sources/v1beta1/kafka_defaults.go
index c767f9c1f..490a7fdb1 100644
--- a/pkg/apis/sources/v1beta1/kafka_defaults.go
+++ b/pkg/apis/sources/v1beta1/kafka_defaults.go
@@ -20,6 +20,7 @@ import (
 	"context"
 
 	"github.com/google/uuid"
+	"k8s.io/utils/pointer"
 )
 
 const (
@@ -28,7 +29,11 @@ const (
 
 // SetDefaults ensures KafkaSource reflects the default values.
 func (k *KafkaSource) SetDefaults(ctx context.Context) {
-	if k != nil && k.Spec.ConsumerGroup == "" {
+	if k.Spec.ConsumerGroup == "" {
 		k.Spec.ConsumerGroup = uuidPrefix + uuid.New().String()
 	}
+
+	if k.Spec.Consumers == nil {
+		k.Spec.Consumers = pointer.Int32Ptr(1)
+	}
 }


Test patch:
diff --git a/pkg/apis/messaging/v1alpha1/kafka_channel_defaults_test.go b/pkg/apis/messaging/v1alpha1/kafka_channel_defaults_test.go
index f5151a6fb..c897c4d9c 100644
--- a/pkg/apis/messaging/v1alpha1/kafka_channel_defaults_test.go
+++ b/pkg/apis/messaging/v1alpha1/kafka_channel_defaults_test.go
@@ -36,7 +36,7 @@ func TestKafkaChannelDefaults(t *testing.T) {
 			initial: KafkaChannel{},
 			expected: KafkaChannel{
 				ObjectMeta: metav1.ObjectMeta{
-					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1alpha1"},
+					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1beta1"},
 				},
 				Spec: KafkaChannelSpec{
 					NumPartitions:     constants.DefaultNumPartitions,
@@ -52,7 +52,7 @@ func TestKafkaChannelDefaults(t *testing.T) {
 			},
 			expected: KafkaChannel{
 				ObjectMeta: metav1.ObjectMeta{
-					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1alpha1"},
+					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1beta1"},
 				},
 				Spec: KafkaChannelSpec{
 					NumPartitions:     constants.DefaultNumPartitions,
@@ -68,7 +68,7 @@ func TestKafkaChannelDefaults(t *testing.T) {
 			},
 			expected: KafkaChannel{
 				ObjectMeta: metav1.ObjectMeta{
-					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1alpha1"},
+					Annotations: map[string]string{"messaging.knative.dev/subscribable": "v1beta1"},
 				},
 				Spec: KafkaChannelSpec{
 					NumPartitions:     testNumPartitions,
diff --git a/pkg/apis/sources/v1beta1/kafka_defaults_test.go b/pkg/apis/sources/v1beta1/kafka_defaults_test.go
index 4bd1ec1f5..275a00aa6 100644
--- a/pkg/apis/sources/v1beta1/kafka_defaults_test.go
+++ b/pkg/apis/sources/v1beta1/kafka_defaults_test.go
@@ -18,11 +18,13 @@ package v1beta1
 
 import (
 	"context"
+	"strconv"
 	"strings"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
+	"k8s.io/utils/pointer"
 )
 
 type defaultKafkaTestArgs struct {
@@ -45,6 +47,13 @@ func TestSetDefaults(t *testing.T) {
 			t.Fatalf("Unexpected consumerGroup Set (-want, +got): %s", diff)
 		}
 	}
+	assertConsumers := func(t *testing.T, ks KafkaSource, expected string) {
+		i, _ := strconv.Atoi(expected)
+		i32 := int32(i)
+		if diff := cmp.Diff(ks.Spec.Consumers, &i32); diff != "" {
+			t.Fatalf("Unexpected consumers (-want, +got): %s", diff)
+		}
+	}
 	testCases := []defaultKafkaTestArgs{
 		{
 			Name:       "nil spec",
@@ -61,6 +70,18 @@ func TestSetDefaults(t *testing.T) {
 			Expected:   "foo",
 			AssertFunc: assertGivenGroup,
 		},
+		{
+			Name:       "consumers not set",
+			Initial:    KafkaSource{},
+			Expected:   "1",
+			AssertFunc: assertConsumers,
+		},
+		{
+			Name:       "consumers set",
+			Initial:    KafkaSource{Spec: KafkaSourceSpec{Consumers: pointer.Int32Ptr(4)}},
+			Expected:   "4",
+			AssertFunc: assertConsumers,
+		},
 	}
 
 	for _, tc := range testCases {