Repository: knative-sandbox/eventing-kafka-broker
Pull request #3157 (2023-06-19 20:54:45)

Allow defining topic prefixes
<!-- Please include the 'why' behind your changes if no issue exists -->
As part of #3153 we need to be able to add broker and channel topic prefixes.
## Proposed Changes

- Update the kafka features to support broker topic template and channel topic template


knative-prow[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/3157#" title="Author self-approved">Cali0707</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign dprotaso for approval. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["dprotaso"]} -->
Cali0707: /cc @matzew 
knative-prow[bot]: @Cali0707: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_main | 5d1ea4fb16448d8d24cd0e701633c57236636c62 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3157/unit-tests_eventing-kafka-broker_main/1670897938492035073) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
Cali0707: /retest-required
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/3157?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#3157](https://app.codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/3157?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (5d1ea4f) into [main](https://app.codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/a0068915df8e3c6700852b4f3cdab8a64d8cc36c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (a006891) will **decrease** coverage by `5.15%`.
> The diff coverage is `83.33%`.

```diff
@@             Coverage Diff              @@
##               main    #3157      +/-   ##
============================================
- Coverage     63.54%   58.39%   -5.15%     
============================================
  Files           167       90      -77     
  Lines         11506     8878    -2628     
  Branches        233        0     -233     
============================================
- Hits           7311     5184    -2127     
+ Misses         3656     3291     -365     
+ Partials        539      403     -136     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://app.codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/3157?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/apis/config/features.go](https://app.codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/3157?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9jb25maWcvZmVhdHVyZXMuZ28=) | `86.31% <83.33%> (+1.97%)` | :arrow_up: |

... and [77 files with indirect coverage changes](https://app.codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/3157/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)


knative-prow[bot]: @Cali0707: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | 5d1ea4fb16448d8d24cd0e701633c57236636c62 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3157/channel-integration-tests-ssl_eventing-kafka-broker_main/1670897938794024960) | true | `/test channel-integration-tests-ssl`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @Cali0707: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | 5d1ea4fb16448d8d24cd0e701633c57236636c62 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3157/channel-integration-tests-ssl_eventing-kafka-broker_main/1670897938794024960) | true | `/test channel-integration-tests-ssl`
unit-tests_eventing-kafka-broker_main | 2c9f35a7d7e65bd5135048215200904c812e1a10 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3157/unit-tests_eventing-kafka-broker_main/1671145047732523008) | true | `/test unit-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: @Cali0707: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_main | 5d1ea4fb16448d8d24cd0e701633c57236636c62 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3157/channel-integration-tests-ssl_eventing-kafka-broker_main/1670897938794024960) | true | `/test channel-integration-tests-ssl`
unit-tests_eventing-kafka-broker_main | 2c9f35a7d7e65bd5135048215200904c812e1a10 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3157/unit-tests_eventing-kafka-broker_main/1671145047732523008) | true | `/test unit-tests`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 2c9f35a7d7e65bd5135048215200904c812e1a10 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/3157/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1671145048382640128) | true | `/test channel-integration-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/3157#" title="Author self-approved">Cali0707</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/3157#discussion_r1235440743" title="Approved">pierDipi</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)~~ [pierDipi]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml
index 7a260ab78..df8780a26 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml
@@ -4,7 +4,7 @@ metadata:
   name: config-kafka-features
   namespace: knative-eventing
   annotations:
-    knative.dev/example-checksum: "330d9f60"
+    knative.dev/example-checksum: "1192895d"
 data:
   _example: |-
     ################################
@@ -36,7 +36,15 @@ data:
     # The Go text/template used to generate consumergroup ID for triggers.
     # The template can reference the trigger Kubernetes metadata only.
     triggers.consumergroup.template: "knative-trigger-{{ .Namespace }}-{{ .Name }}"
+    # The Go text/template used to generate topics for Brokers.
+    # The template can reference the broker Kubernetes metadata only.
+    brokers.topic.template: "knative-broker-{{ .Namespace }}-{{ .Name }}"
+    # The Go text/template used to generate topics for Channels.
+    # The template can reference the channel Kubernetes metadata only.
+    channels.topic.template: "knative-channel-{{ .Namespace }}-{{ .Name }}"
   dispatcher.rate-limiter: "disabled"
   dispatcher.ordered-executor-metrics: "disabled"
   controller.autoscaler: "disabled"
   triggers.consumergroup.template: "knative-trigger-{{ .Namespace }}-{{ .Name }}"
+  brokers.topic.template: "knative-broker-{{ .Namespace }}-{{ .Name }}"
+  channels.topic.template: "knative-channel-{{ .Namespace }}-{{ .Name }}"
diff --git a/control-plane/pkg/apis/config/features.go b/control-plane/pkg/apis/config/features.go
index b416b75e1..34032d6fc 100644
--- a/control-plane/pkg/apis/config/features.go
+++ b/control-plane/pkg/apis/config/features.go
@@ -40,6 +40,8 @@ type features struct {
 	DispatcherOrderedExecutorMetrics feature.Flag
 	ControllerAutoscaler             feature.Flag
 	TriggersConsumerGroupTemplate    *template.Template
+	BrokersTopicTemplate             *template.Template
+	ChannelsTopicTemplate            *template.Template
 }
 
 type KafkaFeatureFlags struct {
@@ -47,10 +49,16 @@ type KafkaFeatureFlags struct {
 	m        sync.RWMutex
 }
 
-var DefaultTriggersConsumerGroupTemplate *template.Template
+var (
+	DefaultTriggersConsumerGroupTemplate *template.Template
+	DefaultBrokersTopicTemplate          *template.Template
+	DefaultChannelsTopicTemplate         *template.Template
+)
 
 func init() {
 	DefaultTriggersConsumerGroupTemplate, _ = template.New("triggers.consumergroup.template").Parse("knative-trigger-{{ .Namespace }}-{{ .Name }}")
+	DefaultBrokersTopicTemplate, _ = template.New("brokers.topic.template").Parse("knative-broker-{{ .Namespace }}-{{ .Name }}")
+	DefaultChannelsTopicTemplate, _ = template.New("channels.topic.template").Parse("knative-channel-{{ .Namespace }}-{{ .Name }}")
 }
 
 func DefaultFeaturesConfig() *KafkaFeatureFlags {
@@ -60,6 +68,8 @@ func DefaultFeaturesConfig() *KafkaFeatureFlags {
 			DispatcherOrderedExecutorMetrics: feature.Disabled,
 			ControllerAutoscaler:             feature.Disabled,
 			TriggersConsumerGroupTemplate:    DefaultTriggersConsumerGroupTemplate,
+			BrokersTopicTemplate:             DefaultBrokersTopicTemplate,
+			ChannelsTopicTemplate:            DefaultChannelsTopicTemplate,
 		},
 	}
 }
@@ -72,6 +82,8 @@ func newFeaturesConfigFromMap(cm *corev1.ConfigMap) (*KafkaFeatureFlags, error)
 		asFlag("dispatcher.ordered-executor-metrics", &nc.features.DispatcherOrderedExecutorMetrics),
 		asFlag("controller.autoscaler", &nc.features.ControllerAutoscaler),
 		asTemplate("triggers.consumergroup.template", nc.features.TriggersConsumerGroupTemplate),
+		asTemplate("brokers.topic.template", nc.features.BrokersTopicTemplate),
+		asTemplate("channels.topic.template", nc.features.ChannelsTopicTemplate),
 	)
 	return nc, err
 }
@@ -95,13 +107,15 @@ func (f *KafkaFeatureFlags) IsControllerAutoscalerEnabled() bool {
 }
 
 func (f *KafkaFeatureFlags) ExecuteTriggersConsumerGroupTemplate(triggerMetadata v1.ObjectMeta) (string, error) {
-	var result bytes.Buffer
-	err := f.features.TriggersConsumerGroupTemplate.Execute(&result, triggerMetadata)
-	if err != nil {
-		return "", fmt.Errorf("unable to execute triggers consumergroup template: %w", err)
-	}
+	return executeTemplateToString(f.features.TriggersConsumerGroupTemplate, triggerMetadata, "unable to execute triggers consumergroup template: %w")
+}
 
-	return result.String(), nil
+func (f *KafkaFeatureFlags) ExecuteBrokersTopicTemplate(brokerMetadata v1.ObjectMeta) (string, error) {
+	return executeTemplateToString(f.features.BrokersTopicTemplate, brokerMetadata, "unable to execute brokers topic template: %w")
+}
+
+func (f *KafkaFeatureFlags) ExecuteChannelsTopicTemplate(channelMetadata v1.ObjectMeta) (string, error) {
+	return executeTemplateToString(f.features.ChannelsTopicTemplate, channelMetadata, "unable to execute channels topic template: %w")
 }
 
 // Store is a typed wrapper around configmap.Untyped store to handle our configmaps.
@@ -186,3 +200,13 @@ func asTemplate(key string, target *template.Template) configmap.ParseFunc {
 		return nil
 	}
 }
+
+func executeTemplateToString(template *template.Template, metadata v1.ObjectMeta, errorMessage string) (string, error) {
+	var result bytes.Buffer
+	err := template.Execute(&result, metadata)
+	if err != nil {
+		return "", fmt.Errorf(errorMessage, err)
+	}
+
+	return result.String(), nil
+}


Test patch:
diff --git a/control-plane/pkg/apis/config/features_test.go b/control-plane/pkg/apis/config/features_test.go
index 37a5514b5..dc0860013 100644
--- a/control-plane/pkg/apis/config/features_test.go
+++ b/control-plane/pkg/apis/config/features_test.go
@@ -60,6 +60,11 @@ func TestGetFlags(t *testing.T) {
 	require.True(t, flags.IsControllerAutoscalerEnabled())
 	require.Len(t, flags.features.TriggersConsumerGroupTemplate.Tree.Root.Nodes, 4)
 	require.Equal(t, flags.features.TriggersConsumerGroupTemplate.Name(), "triggers.consumergroup.template")
+	require.Len(t, flags.features.BrokersTopicTemplate.Tree.Root.Nodes, 4)
+	require.Equal(t, flags.features.BrokersTopicTemplate.Name(), "brokers.topic.template")
+	require.Len(t, flags.features.ChannelsTopicTemplate.Tree.Root.Nodes, 4)
+	require.Equal(t, flags.features.ChannelsTopicTemplate.Name(), "channels.topic.template")
+
 }
 
 func TestStoreLoadWithConfigMap(t *testing.T) {
@@ -75,6 +80,8 @@ func TestStoreLoadWithConfigMap(t *testing.T) {
 	require.Equal(t, expected.IsDispatcherOrderedExecutorMetricsEnabled(), have.IsDispatcherOrderedExecutorMetricsEnabled())
 	require.Equal(t, expected.IsControllerAutoscalerEnabled(), have.IsControllerAutoscalerEnabled())
 	require.Equal(t, expected.features.TriggersConsumerGroupTemplate.Name(), have.features.TriggersConsumerGroupTemplate.Name())
+	require.Equal(t, expected.features.BrokersTopicTemplate.Name(), have.features.BrokersTopicTemplate.Name())
+	require.Equal(t, expected.features.ChannelsTopicTemplate.Name(), have.features.ChannelsTopicTemplate.Name())
 }
 
 func TestStoreLoadWithContext(t *testing.T) {
@@ -84,6 +91,8 @@ func TestStoreLoadWithContext(t *testing.T) {
 	require.False(t, have.IsDispatcherOrderedExecutorMetricsEnabled())
 	require.False(t, have.IsControllerAutoscalerEnabled())
 	require.Equal(t, have.features.TriggersConsumerGroupTemplate.Name(), "triggers.consumergroup.template")
+	require.Equal(t, have.features.BrokersTopicTemplate.Name(), "brokers.topic.template")
+	require.Equal(t, have.features.ChannelsTopicTemplate.Name(), "channels.topic.template")
 }
 
 func TestExecuteTriggersConsumerGroupTemplateDefault(t *testing.T) {
@@ -115,3 +124,63 @@ func TestExecuteTriggersConsumerGroupTemplateOverride(t *testing.T) {
 
 	require.Equal(t, result, "knative-trigger-namespace-trigger-138ac0ec-2694-4747-900d-45be3da5c9a9")
 }
+
+func TestExecuteBrokersTopicTemplateDefault(t *testing.T) {
+	nc := DefaultFeaturesConfig()
+	result, err := nc.ExecuteBrokersTopicTemplate(v1.ObjectMeta{
+		Name:      "topic",
+		Namespace: "namespace",
+		UID:       "138ac0ec-2694-4747-900d-45be3da5c9a9",
+	})
+	if err != nil {
+		require.NoError(t, err)
+	}
+
+	require.Equal(t, result, "knative-broker-namespace-topic")
+}
+
+func TestExecuteBrokersTopicTemplateOverride(t *testing.T) {
+	nc := DefaultFeaturesConfig()
+	nc.features.BrokersTopicTemplate, _ = template.New("custom-template").Parse("knative-broker-{{ .Namespace }}-{{ .Name }}-{{ .UID }}")
+
+	result, err := nc.ExecuteBrokersTopicTemplate(v1.ObjectMeta{
+		Name:      "topic",
+		Namespace: "namespace",
+		UID:       "138ac0ec-2694-4747-900d-45be3da5c9a9",
+	})
+	if err != nil {
+		require.NoError(t, err)
+	}
+
+	require.Equal(t, result, "knative-broker-namespace-topic-138ac0ec-2694-4747-900d-45be3da5c9a9")
+}
+
+func TestExecuteChannelsTopicTemplateDefault(t *testing.T) {
+	nc := DefaultFeaturesConfig()
+	result, err := nc.ExecuteChannelsTopicTemplate(v1.ObjectMeta{
+		Name:      "topic",
+		Namespace: "namespace",
+		UID:       "138ac0ec-2694-4747-900d-45be3da5c9a9",
+	})
+	if err != nil {
+		require.NoError(t, err)
+	}
+
+	require.Equal(t, result, "knative-channel-namespace-topic")
+}
+
+func TestExecuteChannelsTopicTemplateOverride(t *testing.T) {
+	nc := DefaultFeaturesConfig()
+	nc.features.ChannelsTopicTemplate, _ = template.New("custom-template").Parse("knative-channel-{{ .Namespace }}-{{ .Name }}-{{ .UID }}")
+
+	result, err := nc.ExecuteChannelsTopicTemplate(v1.ObjectMeta{
+		Name:      "topic",
+		Namespace: "namespace",
+		UID:       "138ac0ec-2694-4747-900d-45be3da5c9a9",
+	})
+	if err != nil {
+		require.NoError(t, err)
+	}
+
+	require.Equal(t, result, "knative-channel-namespace-topic-138ac0ec-2694-4747-900d-45be3da5c9a9")
+}
diff --git a/control-plane/pkg/apis/config/testdata/config-kafka-features.yaml b/control-plane/pkg/apis/config/testdata/config-kafka-features.yaml
index d0495bc4c..5f99736b6 100644
--- a/control-plane/pkg/apis/config/testdata/config-kafka-features.yaml
+++ b/control-plane/pkg/apis/config/testdata/config-kafka-features.yaml
@@ -9,3 +9,5 @@ data:
     dispatcher.ordered-executor-metrics: "enabled"
     controller.autoscaler: "enabled"
     triggers.consumergroup.template: "knative-trigger-{{ .Namespace }}-{{ .Name }}"
+    brokers.topic.template: "knative-broker-{{ .Namespace }}-{{ .Name }}"
+    channels.topic.template: "knative-channel-{{ .Namespace }}-{{ .Name }}"
diff --git a/test/config/100-config-kafka-features.yaml b/test/config/100-config-kafka-features.yaml
index 83765063a..57365e5f9 100644
--- a/test/config/100-config-kafka-features.yaml
+++ b/test/config/100-config-kafka-features.yaml
@@ -8,3 +8,5 @@ data:
   dispatcher.ordered-executor-metrics: "enabled"
   controller.autoscaler: "enabled"
   triggers.consumergroup.template: "knative-trigger-{{ .Namespace }}-{{ .Name }}"
+  brokers.topic.template: "knative-broker-{{ .Namespace }}-{{ .Name }}"
+  channels.topic.template: "knative-channel-{{ .Namespace }}-{{ .Name }}"