Update Broker and Controller to use topic name templates

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

@@ -58,7 +58,8 @@ var (
 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 }}")
+	// 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 {
@@ -74,8 +75,8 @@ func DefaultFeaturesConfig() *KafkaFeatureFlags {
 	}
 }
 
-// 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 +132,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())

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, err := r.KafkaFeatureFlags.ExecuteBrokersTopicTemplate(broker.ObjectMeta)
+	if err != nil {
+		return err
+	}
+	topic, err := kafka.DeleteTopic(kafkaClusterAdminClient, topicName)
 	if err != nil {
 		return err
 	}

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

@@ -21,13 +21,16 @@ import (
 	"fmt"
 	"net/url"
 	"testing"
+	"text/template"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/counter"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	"k8s.io/utils/pointer"
 
+	cm "knative.dev/pkg/configmap/testing"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
@@ -37,11 +40,13 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	clientgotesting "k8s.io/client-go/testing"
+	apisconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/eventing/pkg/eventingtls/eventingtlstesting"
@@ -72,6 +77,10 @@ const (
 	ExpectedTopicDetail    = "expectedTopicDetail"
 	testProber             = "testProber"
 	externalTopic          = "externalTopic"
+
+	useCustomTopicTemplate = "use-custom-topic-template"
+
+	FlagsConfigName = "config-kafka-features"
 )
 
 const (
@@ -98,8 +107,11 @@ var (
 	createTopicError = fmt.Errorf("failed to create topic")
 	deleteTopicError = fmt.Errorf("failed to delete topic")
 
-	linear      = eventingduck.BackoffPolicyLinear
-	exponential = eventingduck.BackoffPolicyExponential
+	linear            = eventingduck.BackoffPolicyLinear
+	exponential       = eventingduck.BackoffPolicyExponential
+	kafkaFeatureFlags = apisconfig.DefaultFeaturesConfig()
+
+	customBrokerTopicTemplate = customTemplate()
 )
 
 var DefaultEnv = &config.Env{
@@ -2135,6 +2147,81 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
+		}, {
+			Name: "Reconciled normal - with custom topic template",
+			Objects: []runtime.Object{
+				NewBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
+				NewService(),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              BrokerUUID,
+							Topics:           []string{CustomBrokerTopic(customBrokerTopicTemplate)},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
+							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
+						},
+					},
+					Generation: 1,
+				}),
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						reconcilertesting.WithInitBrokerConditions,
+						StatusBrokerConfigMapUpdatedReady(&env),
+						StatusBrokerDataPlaneAvailable,
+						StatusBrokerConfigParsed,
+						StatusExternalBrokerTopicReady(CustomBrokerTopic(customBrokerTopicTemplate)),
+						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
+						BrokerConfigMapAnnotations(),
+						WithTopicStatusAnnotation(CustomBrokerTopic(customBrokerTopicTemplate)),
+						WithBrokerAddresses([]duckv1.Addressable{
+							{
+								Name: pointer.String("http"),
+								URL:  brokerAddress,
+							},
+						}),
+						WithBrokerAddress(duckv1.Addressable{
+							Name: pointer.String("http"),
+							URL:  brokerAddress,
+						}),
+						WithBrokerAddessable(),
+					),
+				},
+			},
+
+			OtherTestData: map[string]interface{}{
+				useCustomTopicTemplate: true,
+			},
 		},
 	}
 
@@ -2774,6 +2861,16 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 			ReplicationFactor: DefaultReplicationFactor,
 		}
 
+		var featureFlags *apisconfig.KafkaFeatureFlags
+		var err error
+		if useCustomTemplate, ok := row.OtherTestData[useCustomTopicTemplate]; ok == true && useCustomTemplate == true {
+			_, example := cm.ConfigMapsFromTestFile(t, FlagsConfigName)
+			featureFlags, err = apisconfig.NewFeaturesConfigFromMap(example)
+			require.NoError(t, err)
+		} else {
+			featureFlags = apisconfig.DefaultFeaturesConfig()
+		}
+
 		var onCreateTopicError error
 		if want, ok := row.OtherTestData[wantErrorOnCreateTopic]; ok {
 			onCreateTopicError = want.(error)
@@ -2789,7 +2886,8 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 			expectedTopicDetail = td.(sarama.TopicDetail)
 		}
 
-		expectedTopicName := fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName)
+		expectedTopicName, err := featureFlags.ExecuteBrokersTopicTemplate(metav1.ObjectMeta{Namespace: BrokerNamespace, Name: BrokerName, UID: BrokerUUID})
+		require.NoError(t, err, "Failed to create broker topic name from feature flags")
 		if t, ok := row.OtherTestData[externalTopic]; ok {
 			expectedTopicName = t.(string)
 		}
@@ -2830,9 +2928,10 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 					T:                                      t,
 				}, nil
 			},
-			Env:     env,
-			Prober:  proberMock,
-			Counter: counter.NewExpiringCounter(ctx),
+			Env:               env,
+			Prober:            proberMock,
+			Counter:           counter.NewExpiringCounter(ctx),
+			KafkaFeatureFlags: featureFlags,
 		}
 
 		reconciler.Tracker = &FakeTracker{}
@@ -2883,3 +2982,8 @@ func httpsURL(name string, namespace string) *apis.URL {
 		Path:   fmt.Sprintf("/%s/%s", namespace, name),
 	}
 }
+
+func customTemplate() *template.Template {
+	brokersTemplate, _ := template.New("brokers.topic.template").Parse("custom-broker-template.{{ .Namespace }}-{{ .Name }}")
+	return brokersTemplate
+}

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

@@ -39,6 +39,7 @@ import (
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 
+	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/counter"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
@@ -56,6 +57,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
 
 	configmapInformer := configmapinformer.Get(ctx)
+	featureFlags := apisconfig.DefaultFeaturesConfig()
 
 	reconciler := &Reconciler{
 		Reconciler: &base.Reconciler{
@@ -73,6 +75,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 		ConfigMapLister:            configmapInformer.Lister(),
 		Env:                        env,
 		Counter:                    counter.NewExpiringCounter(ctx),
+		KafkaFeatureFlags:          featureFlags,
 	}
 
 	logger := logging.FromContext(ctx)

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

@@ -46,6 +46,7 @@ import (
 	brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 
+	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/counter"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
@@ -86,6 +87,8 @@ type NamespacedReconciler struct {
 	ManifestivalClient mf.Client
 
 	DataplaneLifecycleLocksByNamespace util.LockMap[string]
+
+	KafkaFeatureFlags *apisconfig.KafkaFeatureFlags
 }
 
 func (r *NamespacedReconciler) ReconcileKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
@@ -238,6 +241,7 @@ func (r *NamespacedReconciler) createReconcilerForBrokerInstance(broker *eventin
 		BootstrapServers:           r.BootstrapServers,
 		Prober:                     r.Prober,
 		Counter:                    r.Counter,
+		KafkaFeatureFlags:          r.KafkaFeatureFlags,
 	}
 }
 

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

@@ -40,6 +40,7 @@ import (
 
 	"github.com/Shopify/sarama"
 	"github.com/manifestival/client-go-client"
+	"github.com/stretchr/testify/require"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -59,6 +60,7 @@ import (
 	brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
 	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 
+	apisconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
@@ -452,7 +454,8 @@ func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
 			expectedTopicDetail = td.(sarama.TopicDetail)
 		}
 
-		expectedTopicName := fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName)
+		expectedTopicName, err := kafkaFeatureFlags.ExecuteBrokersTopicTemplate(metav1.ObjectMeta{Namespace: BrokerNamespace, Name: BrokerName})
+		require.NoError(t, err, "Failed to create broker topic name from feature flags")
 		if t, ok := row.OtherTestData[externalTopic]; ok {
 			expectedTopicName = t.(string)
 		}
@@ -506,6 +509,7 @@ func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
 			Prober:                             proberMock,
 			ManifestivalClient:                 mfcMockClient,
 			DataplaneLifecycleLocksByNamespace: util.NewExpiringLockMap[string](ctx, time.Minute*30),
+			KafkaFeatureFlags:                  apisconfig.DefaultFeaturesConfig(),
 		}
 
 		r := brokerreconciler.NewReconciler(

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

@@ -56,6 +56,7 @@ import (
 	serviceaccountinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/serviceaccount"
 	clusterrolebindinginformer "knative.dev/pkg/client/injection/kube/informers/rbac/v1/clusterrolebinding"
 
+	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/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
@@ -105,6 +106,7 @@ func NewNamespacedController(ctx context.Context, watcher configmap.Watcher, env
 		Counter:                            counter.NewExpiringCounter(ctx),
 		ManifestivalClient:                 mfc,
 		DataplaneLifecycleLocksByNamespace: util.NewExpiringLockMap[string](ctx, time.Minute*30),
+		KafkaFeatureFlags:                  apisconfig.DefaultFeaturesConfig(),
 	}
 
 	impl := brokerreconciler.NewImpl(ctx, reconciler, kafka.NamespacedBrokerClass, func(impl *controller.Impl) controller.Options {