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

Issue:
Allow specifying a user-provided topic and disable topic creation
**Problem**

- https://github.com/knative-sandbox/eventing-kafka/issues/233

**[Persona:](https://github.com/knative/eventing/blob/master/docs/personas.md)** 
Which persona is this feature for? 

Event producer and cosumer

**Exit Criteria**
A measurable (binary) test that would indicate that the problem has been resolved.

Allow specifying a user-provided topic and the broker is reconciled using credentials without topic create permission.

**Time Estimate (optional):** 
How many developer-days do you think this may take to resolve?
1

**Additional context (optional)**

- This will allow lowering our required permissions to use the Broker.
- This must follow the experimental feature process


Pull request description:
Adding annotation to drive the name of the backing topic
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

Fixes #977 

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- adding support for `kafka.eventing.knative.dev/external.topic` annotation for external topic usage

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: 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
With the `kafka.eventing.knative.dev/external.topic` annotation it is possible to use an externally managed Apache Kafka topic for the broker usage
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


Patch:
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 50665fc07..dfeb9cbf9 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -48,6 +48,9 @@ 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"
 )
 
 type Reconciler struct {
@@ -227,26 +230,42 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 
 func (r *Reconciler) reconcileBrokerTopic(broker *eventing.Broker, securityOption kafka.ConfigOption, statusConditionManager base.StatusConditionManager, topicConfig *kafka.TopicConfig, logger *zap.Logger) (string, reconciler.Event) {
 
-	topicName := kafka.BrokerTopic(TopicPrefix, broker)
 	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 	if err != nil {
-		return "", statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("error getting cluster admin config: %w", err))
+		return "", statusConditionManager.FailedToResolveConfig(fmt.Errorf("error getting cluster admin config: %w", err))
 	}
 
 	kafkaClusterAdminClient, err := r.NewKafkaClusterAdminClient(topicConfig.BootstrapServers, saramaConfig)
 	if err != nil {
-		return "", statusConditionManager.FailedToCreateTopic(topicName, fmt.Errorf("cannot obtain Kafka cluster admin, %w", err))
+		return "", statusConditionManager.FailedToResolveConfig(fmt.Errorf("cannot obtain Kafka cluster admin, %w", err))
 	}
 	defer kafkaClusterAdminClient.Close()
 
-	topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, topicName, topicConfig)
-	if err != nil {
-		return "", statusConditionManager.FailedToCreateTopic(topic, err)
+	// if we have a custom topic annotation
+	// the topic is externally manged and we do NOT need to create it
+	topicName, externalTopic := isExternalTopic(broker)
+	if externalTopic {
+		isPresentAndValid, err := kafka.AreTopicsPresentAndValid(kafkaClusterAdminClient, topicName)
+		if err != nil {
+			return "", statusConditionManager.TopicsNotPresentOrInvalidErr([]string{topicName}, err)
+		}
+		if !isPresentAndValid {
+			// The topic might be invalid.
+			return "", statusConditionManager.TopicsNotPresentOrInvalid([]string{topicName})
+		}
+	} else {
+		// no external topic, we create it
+		topicName = kafka.BrokerTopic(TopicPrefix, broker)
+
+		topic, err := kafka.CreateTopicIfDoesntExist(kafkaClusterAdminClient, logger, topicName, topicConfig)
+		if err != nil {
+			return "", statusConditionManager.FailedToCreateTopic(topic, err)
+		}
 	}
 
-	statusConditionManager.TopicReady(topic)
-	logger.Debug("Topic created", zap.Any("topic", topic))
-	return topic, nil
+	statusConditionManager.TopicReady(topicName)
+	logger.Debug("Topic created", zap.Any("topic", topicName))
+	return topicName, nil
 }
 
 func (r *Reconciler) FinalizeKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
@@ -313,34 +332,38 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 		return controller.NewRequeueAfter(5 * time.Second)
 	}
 
-	topicConfig, brokerConfig, err := r.topicConfig(logger, broker)
-	if err != nil {
-		return fmt.Errorf("failed to resolve broker config: %w", err)
-	}
+	// External topics are not managed by the broker,
+	// therefore we do not delete them
+	_, externalTopic := isExternalTopic(broker)
+	if !externalTopic {
+		topicConfig, brokerConfig, err := r.topicConfig(logger, broker)
+		if err != nil {
+			return fmt.Errorf("failed to resolve broker config: %w", err)
+		}
 
-	secret, err := security.Secret(ctx, &security.MTConfigMapSecretLocator{ConfigMap: brokerConfig, UseNamespaceInConfigmap: false}, r.SecretProviderFunc())
-	if err != nil {
-		return fmt.Errorf("failed to get secret: %w", err)
-	}
-	if secret != nil {
-		logger.Debug("Secret reference",
-			zap.String("apiVersion", secret.APIVersion),
-			zap.String("name", secret.Name),
-			zap.String("namespace", secret.Namespace),
-			zap.String("kind", secret.Kind),
-		)
-	}
+		secret, err := security.Secret(ctx, &security.MTConfigMapSecretLocator{ConfigMap: brokerConfig, UseNamespaceInConfigmap: false}, r.SecretProviderFunc())
+		if err != nil {
+			return fmt.Errorf("failed to get secret: %w", err)
+		}
+		if secret != nil {
+			logger.Debug("Secret reference",
+				zap.String("apiVersion", secret.APIVersion),
+				zap.String("name", secret.Name),
+				zap.String("namespace", secret.Namespace),
+				zap.String("kind", secret.Kind),
+			)
+		}
 
-	// get security option for Sarama with secret info in it
-	securityOption := security.NewSaramaSecurityOptionFromSecret(secret)
-	if err := r.finalizeBrokerTopic(broker, securityOption, topicConfig, logger); err != nil {
-		return err
+		// get security option for Sarama with secret info in it
+		securityOption := security.NewSaramaSecurityOptionFromSecret(secret)
+		if err := r.finalizeNonExternalBrokerTopic(broker, securityOption, topicConfig, logger); err != nil {
+			return err
+		}
 	}
-
 	return nil
 }
 
-func (r *Reconciler) finalizeBrokerTopic(broker *eventing.Broker, securityOption kafka.ConfigOption, topicConfig *kafka.TopicConfig, logger *zap.Logger) reconciler.Event {
+func (r *Reconciler) finalizeNonExternalBrokerTopic(broker *eventing.Broker, securityOption kafka.ConfigOption, topicConfig *kafka.TopicConfig, logger *zap.Logger) reconciler.Event {
 	saramaConfig, err := kafka.GetSaramaConfig(securityOption)
 	if err != nil {
 		// even in error case, we return `normal`, since we are fine with leaving the
@@ -468,3 +491,8 @@ func (r *Reconciler) reconcilerBrokerResource(ctx context.Context, topic string,
 
 	return resource, nil
 }
+
+func isExternalTopic(broker *eventing.Broker) (string, bool) {
+	topicAnnotationValue, ok := broker.Annotations[ExternalTopicAnnotation]
+	return topicAnnotationValue, ok
+}


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index a6dcec5e2..4b53865b4 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -65,6 +65,7 @@ const (
 	wantErrorOnDeleteTopic = "wantErrorOnDeleteTopic"
 	ExpectedTopicDetail    = "expectedTopicDetail"
 	testProber             = "testProber"
+	externalTopic          = "externalTopic"
 )
 
 const (
@@ -173,6 +174,112 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
+		}, {
+			Name: "Reconciled normal - with external topic",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithExternalTopic(ExternalTopicName),
+				),
+				BrokerConfig(bootstrapServers, 20, 5),
+				NewConfigMapWithBinaryData(&env, 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, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              BrokerUUID,
+							Topics:           []string{ExternalTopicName},
+							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(
+						WithExternalTopic(ExternalTopicName),
+						reconcilertesting.WithInitBrokerConditions,
+						StatusBrokerConfigMapUpdatedReady(&env),
+						StatusBrokerDataPlaneAvailable,
+						StatusBrokerConfigParsed,
+						StatusExternalBrokerTopicReady(ExternalTopicName),
+						BrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
+						BrokerConfigMapAnnotations(),
+					),
+				},
+			},
+
+			OtherTestData: map[string]interface{}{
+				externalTopic: ExternalTopicName,
+			},
+		},
+		{
+			Name: "external topic not present or invalid",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithExternalTopic("my-not-present-topic"),
+				),
+				BrokerConfig(bootstrapServers, 20, 5),
+				BrokerReceiverPod(env.SystemNamespace, nil),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
+			},
+			Key:     testKey,
+			WantErr: true,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					"topics %v not present or invalid: invalid topic %s",
+					[]string{"my-not-present-topic"}, "my-not-present-topic",
+				),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithExternalTopic("my-not-present-topic"),
+						reconcilertesting.WithInitBrokerConditions,
+						StatusBrokerDataPlaneAvailable,
+						StatusBrokerConfigParsed,
+						StatusExternalBrokerTopicNotPresentOrInvalid("my-not-present-topic"),
+						BrokerConfigMapAnnotations(),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				externalTopic: "my-not-present-topic",
+			},
 		},
 		{
 			Name: "Reconciled failed - probe " + prober.StatusNotReady.String(),
@@ -2085,6 +2192,18 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 			expectedTopicDetail = td.(sarama.TopicDetail)
 		}
 
+		expectedTopicName := fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName)
+		if t, ok := row.OtherTestData[externalTopic]; ok {
+			expectedTopicName = t.(string)
+		}
+
+		var metadata []*sarama.TopicMetadata
+		metadata = append(metadata, &sarama.TopicMetadata{
+			Name:       ExternalTopicName,
+			IsInternal: false,
+			Partitions: []*sarama.PartitionMetadata{{}},
+		})
+
 		proberMock := probertesting.MockProber(prober.StatusReady)
 		if p, ok := row.OtherTestData[testProber]; ok {
 			proberMock = p.(prober.Prober)
@@ -2105,11 +2224,13 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 			ConfigMapLister: listers.GetConfigMapLister(),
 			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
-					ExpectedTopicName:   fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName),
-					ExpectedTopicDetail: expectedTopicDetail,
-					ErrorOnCreateTopic:  onCreateTopicError,
-					ErrorOnDeleteTopic:  onDeleteTopicError,
-					T:                   t,
+					ExpectedTopicName:                      expectedTopicName,
+					ExpectedTopicDetail:                    expectedTopicDetail,
+					ErrorOnCreateTopic:                     onCreateTopicError,
+					ErrorOnDeleteTopic:                     onDeleteTopicError,
+					ExpectedTopics:                         []string{expectedTopicName},
+					ExpectedTopicsMetadataOnDescribeTopics: metadata,
+					T:                                      t,
 				}, nil
 			},
 			Env:         env,
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 2ea14d1f9..408bf34c4 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -43,9 +43,10 @@ import (
 )
 
 const (
-	BrokerUUID      = "e7185016-5d98-4b54-84e8-3b1cd4acc6b4"
-	BrokerNamespace = "test-namespace"
-	BrokerName      = "test-broker"
+	BrokerUUID        = "e7185016-5d98-4b54-84e8-3b1cd4acc6b4"
+	BrokerNamespace   = "test-namespace"
+	BrokerName        = "test-broker"
+	ExternalTopicName = "test-topic"
 
 	TriggerName      = "test-trigger"
 	TriggerNamespace = "test-namespace"
@@ -132,6 +133,17 @@ func WithBrokerConfig(reference *duckv1.KReference) func(*eventing.Broker) {
 	}
 }
 
+func WithExternalTopic(topic string) func(*eventing.Broker) {
+	return func(broker *eventing.Broker) {
+		annotations := broker.GetAnnotations()
+		if annotations == nil {
+			annotations = make(map[string]string, 1)
+		}
+		annotations[ExternalTopicAnnotation] = topic
+		broker.SetAnnotations(annotations)
+	}
+}
+
 type CMOption func(cm *corev1.ConfigMap)
 
 func BrokerConfig(bootstrapServers string, numPartitions, replicationFactor int, options ...CMOption) *corev1.ConfigMap {
@@ -189,6 +201,12 @@ func StatusBrokerTopicReady(broker *eventing.Broker) {
 	StatusTopicReadyWithName(kafka.BrokerTopic(TopicPrefix, broker))(broker)
 }
 
+func StatusExternalBrokerTopicReady(topic string) func(broker *eventing.Broker) {
+	return func(broker *eventing.Broker) {
+		StatusTopicReadyWithName(topic)(broker)
+	}
+}
+
 func StatusBrokerDataPlaneAvailable(broker *eventing.Broker) {
 	StatusDataPlaneAvailable(broker)
 }
@@ -244,9 +262,13 @@ func BrokerDLSResolved(uri string) func(broker *eventing.Broker) {
 }
 
 func StatusBrokerFailedToCreateTopic(broker *eventing.Broker) {
-
 	StatusFailedToCreateTopic(BrokerTopic())(broker)
+}
 
+func StatusExternalBrokerTopicNotPresentOrInvalid(topicname string) func(broker *eventing.Broker) {
+	return func(broker *eventing.Broker) {
+		StatusTopicNotPresentOrInvalid(topicname)(broker)
+	}
 }
 
 func BrokerDispatcherPod(namespace string, annotations map[string]string) runtime.Object {
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index c803e8288..7188ac1e2 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -375,6 +375,16 @@ func StatusFailedToCreateTopic(topicName string) func(obj duckv1.KRShaped) {
 	}
 }
 
+func StatusTopicNotPresentOrInvalid(topicName string) func(obj duckv1.KRShaped) {
+	return func(obj duckv1.KRShaped) {
+		obj.GetConditionSet().Manage(obj.GetStatus()).MarkFalse(
+			base.ConditionTopicReady,
+			base.ReasonTopicNotPresentOrInvalid,
+			fmt.Sprintf("topics %v: invalid topic %s", []string{topicName}, topicName),
+		)
+	}
+}
+
 func StatusInitialOffsetsCommitted(obj duckv1.KRShaped) {
 	obj.GetConditionSet().Manage(obj.GetStatus()).MarkTrue(base.ConditionInitialOffsetsCommitted)
 }