Update Broker and Controller to use topic name templates

control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-features.yaml

@@ -47,4 +47,4 @@ data:
   controller.autoscaler: "disabled"
   triggers.consumergroup.template: "knative-trigger-{{ .Namespace }}-{{ .Name }}"
   brokers.topic.template: "knative-broker-{{ .Namespace }}-{{ .Name }}"
-  channels.topic.template: "knative-channel-{{ .Namespace }}-{{ .Name }}"
+  channels.topic.template: "knative-messaging-kafka.{{ .Namespace }}.{{ .Name }}"

control-plane/pkg/apis/config/features.go

@@ -50,32 +50,47 @@ type KafkaFeatureFlags struct {
 }
 
 var (
-	DefaultTriggersConsumerGroupTemplate *template.Template
-	DefaultBrokersTopicTemplate          *template.Template
-	DefaultChannelsTopicTemplate         *template.Template
+	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 }}")
+	defaultTriggersConsumerGroupTemplate, _ = template.New("triggers.consumergroup.template").Parse("knative-trigger-{{ .Namespace }}-{{ .Name }}")
+	defaultBrokersTopicTemplate, _ = template.New("brokers.topic.template").Parse("knative-broker-{{ .Namespace }}-{{ .Name }}")
+	// This will resolve to the old naming convention, to prevent errors switching over to the new topic templates approach
+	defaultChannelsTopicTemplate, _ = template.New("channels.topic.template").Parse("knative-messaging-kafka.{{ .Namespace }}.{{ .Name }}")
 }
 
 func DefaultFeaturesConfig() *KafkaFeatureFlags {
+	// we need to clone the default values when creating a new features config
+	// otherwise, when calling NewFeaturesConfigFromMap, we will overwrite the default variables
+	triggersGroupTemplate, err := defaultTriggersConsumerGroupTemplate.Clone()
+	if err != nil {
+		panic("failed to clone default triggers group template")
+	}
+	brokersTopicTemplate, err := defaultBrokersTopicTemplate.Clone()
+	if err != nil {
+		panic("failed to clone default brokers topic template")
+	}
+	channelsTopicTemplate, err := defaultChannelsTopicTemplate.Clone()
+	if err != nil {
+		panic("failed to clone default channels topic template")
+	}
 	return &KafkaFeatureFlags{
 		features: features{
 			DispatcherRateLimiter:            feature.Disabled,
 			DispatcherOrderedExecutorMetrics: feature.Disabled,
 			ControllerAutoscaler:             feature.Disabled,
-			TriggersConsumerGroupTemplate:    DefaultTriggersConsumerGroupTemplate,
-			BrokersTopicTemplate:             DefaultBrokersTopicTemplate,
-			ChannelsTopicTemplate:            DefaultChannelsTopicTemplate,
+			TriggersConsumerGroupTemplate:    triggersGroupTemplate,
+			BrokersTopicTemplate:             brokersTopicTemplate,
+			ChannelsTopicTemplate:            channelsTopicTemplate,
 		},
 	}
 }
 
-// newFeaturesConfigFromMap creates a Features from the supplied Map
-func newFeaturesConfigFromMap(cm *corev1.ConfigMap) (*KafkaFeatureFlags, error) {
+// NewFeaturesConfigFromMap creates a Features from the supplied Map
+func NewFeaturesConfigFromMap(cm *corev1.ConfigMap) (*KafkaFeatureFlags, error) {
 	nc := DefaultFeaturesConfig()
 	err := configmap.Parse(cm.Data,
 		asFlag("dispatcher.rate-limiter", &nc.features.DispatcherRateLimiter),
@@ -131,7 +146,7 @@ func NewStore(ctx context.Context, onAfterStore ...func(name string, value *Kafk
 			"config-kafka-features",
 			logging.FromContext(ctx).Named("config-kafka-features"),
 			configmap.Constructors{
-				FlagsConfigName: newFeaturesConfigFromMap,
+				FlagsConfigName: NewFeaturesConfigFromMap,
 			},
 			func(name string, value interface{}) {
 				for _, f := range onAfterStore {

control-plane/pkg/apis/config/features_test.go

@@ -51,7 +51,7 @@ func TestFlags_IsEnabled_ContainingFlag(t *testing.T) {
 
 func TestGetFlags(t *testing.T) {
 	_, example := cm.ConfigMapsFromTestFile(t, FlagsConfigName)
-	flags, err := newFeaturesConfigFromMap(example)
+	flags, err := NewFeaturesConfigFromMap(example)
 	require.NoError(t, err)
 
 	require.True(t, flags.IsDispatcherRateLimiterEnabled())
@@ -74,7 +74,7 @@ func TestStoreLoadWithConfigMap(t *testing.T) {
 	store.OnConfigChanged(exampleConfig)
 
 	have := FromContext(store.ToContext(context.Background()))
-	expected, _ := newFeaturesConfigFromMap(exampleConfig)
+	expected, _ := NewFeaturesConfigFromMap(exampleConfig)
 
 	require.Equal(t, expected.IsDispatcherRateLimiterEnabled(), have.IsDispatcherRateLimiterEnabled())
 	require.Equal(t, expected.IsDispatcherOrderedExecutorMetricsEnabled(), have.IsDispatcherOrderedExecutorMetricsEnabled())
@@ -166,7 +166,7 @@ func TestExecuteChannelsTopicTemplateDefault(t *testing.T) {
 		require.NoError(t, err)
 	}
 
-	require.Equal(t, result, "knative-channel-namespace-topic")
+	require.Equal(t, result, "knative-messaging-kafka.namespace.topic")
 }
 
 func TestExecuteChannelsTopicTemplateOverride(t *testing.T) {

control-plane/pkg/apis/config/testdata/config-kafka-features.yaml

@@ -10,4 +10,4 @@ data:
     controller.autoscaler: "enabled"
     triggers.consumergroup.template: "knative-trigger-{{ .Namespace }}-{{ .Name }}"
     brokers.topic.template: "knative-broker-{{ .Namespace }}-{{ .Name }}"
-    channels.topic.template: "knative-channel-{{ .Namespace }}-{{ .Name }}"
+    channels.topic.template: "knative-messaging-kafka.{{ .Namespace }}.{{ .Name }}"

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

@@ -37,6 +37,7 @@ import (
 	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
 
+	apisconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
@@ -50,9 +51,6 @@ import (
 )
 
 const (
-	// TopicPrefix is the Kafka Broker topic prefix - (topic name: knative-broker-<broker-namespace>-<broker-name>).
-	TopicPrefix = "knative-broker-"
-
 	// ExternalTopicAnnotation for using external kafka topic for the broker
 	ExternalTopicAnnotation = "kafka.eventing.knative.dev/external.topic"
 
@@ -80,8 +78,9 @@ type Reconciler struct {
 
 	BootstrapServers string
 
-	Prober  prober.Prober
-	Counter *counter.Counter
+	Prober            prober.Prober
+	Counter           *counter.Counter
+	KafkaFeatureFlags *apisconfig.KafkaFeatureFlags
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
@@ -311,7 +310,15 @@ func (r *Reconciler) reconcileBrokerTopic(broker *eventing.Broker, securityOptio
 		}
 	} else {
 		// no external topic, we create it
-		topicName = kafka.BrokerTopic(TopicPrefix, broker)
+		var existingTopic bool
+		// check if the broker already has reconciled with a topic name and use that if it exists
+		// otherwise, we create a new topic name from the broker topic template
+		if topicName, existingTopic = broker.Status.Annotations[kafka.TopicAnnotation]; !existingTopic {
+			topicName, err = r.KafkaFeatureFlags.ExecuteBrokersTopicTemplate(broker.ObjectMeta)
+			if err != nil {
+				return "", statusConditionManager.TopicsNotPresentOrInvalidErr([]string{topicName}, err)
+			}
+		}
 
 		topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, topicName, topicConfig)
 		if err != nil {
@@ -503,7 +510,11 @@ func (r *Reconciler) finalizeNonExternalBrokerTopic(broker *eventing.Broker, sec
 	}
 	defer kafkaClusterAdminClient.Close()
 
-	topic, err := kafka.DeleteTopic(kafkaClusterAdminClient, kafka.BrokerTopic(TopicPrefix, broker))
+	topicName, ok := broker.Status.Annotations[kafka.TopicAnnotation]
+	if !ok {
+		return fmt.Errorf("no topic annotated on broker")
+	}
+	topic, err := kafka.DeleteTopic(kafkaClusterAdminClient, topicName)
 	if err != nil {
 		return err
 	}