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

Issue:
[Distributed KafkaChannel] Dispatcher Reconciliation & Reconfigure Lifecycle Problems
**Describe the bug**
There are a couple problems with the Dispatcher's "controller" which is watching KafkaChannels in order to start/stop Kafka ConsumerGroups...

1. The reconciler does not requeue upon reconciliation errors so subscribers can become "stuck" in a bad state until the KafkaChannel is changed.
2. An existing kafkachannel can be put into a not ready state by a bad Reconfigure (Kafka Secret change) and is not returned when the Secret is fixed.
3. A "stopped" subscription would unintentionally be restarted by a Reconfigure (Kafka Secret Change)

 **Expected behavior**
1. The reconciler should return errors to force a requeue of the KafkaChannel in an attempt to resolve the problem.
2. A Subscription is returned to READY state after a bad Kafka Secret change is corrected.
3. A "stoped" subscription should not be affected by a Reconfigure (Kafka Secret Change)

 **To Reproduce**
1. Create a KafkaChannel with a Subscription.  Change the Kafka Secret to invalid values which will trigger a reconfigure which will result in the Subscription being "MarkedNotReadyByChannel".  Verify the reconciler requeues the KafkaChannel and keeps trying (backoff)
2. Create a KafkaChannel with a Subscription.  Change the Kafka Secret to invalid values which will trigger a reconfigure which will result in the Subscription being "MarkedNotReadyByChannel".  Fix the Kafka Secret back to valid values and notice that the Subscription is not returned to a READY state.
3. Really hard to reproduce now since the "Pause" capability is not in place - would have to catch a ResetOffset at just the right moment.  

 **Knative release version**
/main

 **Additional context**
This code needs to respect the control-protocol messaging used to Stop/Start Consumer Groups for the ResetOffset (and eventually the Pause) features.  Subscriptions which are "stopped" should not be restarted as part of a Reconfigure (Kafka Secret Change)


Pull request description:
fix-reconfigure-and-reconciliation-issues
Fixes #794 

## Proposed Changes

- 🧽  Restructure Dispatcher main() to create Controller prior to SecretWatcher.
- 🧽  Change Dispatcher's SecretObserver to be a factory capable of requeueing KafkaChannels.
- 🧽  Add util function to convert from GroupId back to Subscription.UID.
- 🧽  Return errors from Dispatcher Controller Reconcile() to requeue on failure.
- 🧽  Create new ReconfigureError type to contain the failed ConsumerGroup IDs
- 🧽  Dispatcher.SecretChanged() should handle ReconfigureErrors by removing failed ConsumerGroups from Dispatcher.subscribers so that the next reconcile will not ignore them (as being previously closed)
- 🐛 Dispatcher.Reconfigure() bug fixed so that "paused" (stopped) aren't restarted.
- 🧽  Dispatcher.Reconfigure() tracks failed ConsumerGroup IDs and returns new ReconfigureError type.
- 🧽  Removed unneeded DispatcherConfig.SubscriberSpecs

**Release Note**

```release-note
Distributed KafkaChannel Dispatcher will now re-queue KafkaChannels after failed reconciliation which will improve various failure recovery scenarios, but could consume CPU resources if reconciliation is blocked by underlying system issues (bad Kafka Secret config, etc.)
```

Holding for further testing...
/hold

Patch:
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index 05d14f982..5c593bd48 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -24,8 +24,10 @@ import (
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/cache"
 	"knative.dev/eventing/pkg/kncloudevents"
 	injectionclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
@@ -46,19 +48,11 @@ import (
 	kafkaclientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	"knative.dev/eventing-kafka/pkg/client/informers/externalversions"
 	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
-	commonconsumer "knative.dev/eventing-kafka/pkg/common/consumer"
 	"knative.dev/eventing-kafka/pkg/common/controlprotocol"
 	"knative.dev/eventing-kafka/pkg/common/kafka/sarama"
 	"knative.dev/eventing-kafka/pkg/common/metrics"
 )
 
-// Variables
-var (
-	logger        *zap.Logger
-	dispatcher    dispatch.Dispatcher
-	managerEvents <-chan commonconsumer.ManagerEvent
-)
-
 // The Main Function (Go Command)
 func main() {
 
@@ -85,7 +79,7 @@ func main() {
 	ctx = commonk8s.LoggingContext(ctx, constants.Component, k8sClient)
 
 	// Get The Logger From The Context & Defer Flushing Any Buffered Log Entries On Exit
-	logger = logging.FromContext(ctx).Desugar()
+	logger := logging.FromContext(ctx).Desugar()
 	defer flush(logger)
 
 	// Load Environment Variables
@@ -159,37 +153,34 @@ func main() {
 		MetricsRegistry: ekConfig.Sarama.Config.MetricRegistry,
 		SaramaConfig:    ekConfig.Sarama.Config,
 	}
-	dispatcher, managerEvents = dispatch.NewDispatcher(dispatcherConfig, controlProtocolServer)
-
-	// Watch The Secret For Changes
-	err = distributedcommonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, environment.ResyncPeriod, secretObserver)
-	if err != nil {
-		logger.Fatal("Failed To Start Secret Watcher", zap.Error(err))
-	}
+	dispatcher, managerEvents := dispatch.NewDispatcher(dispatcherConfig, controlProtocolServer)
 
+	// Create KafkaChannel Informer
 	kafkaClient := kafkaclientset.NewForConfigOrDie(k8sConfig)
-
 	kafkaInformerFactory := externalversions.NewSharedInformerFactory(kafkaClient, environment.ResyncPeriod)
-
-	// Create KafkaChannel Informer
 	kafkaChannelInformer := kafkaInformerFactory.Messaging().V1beta1().KafkaChannels()
 
-	// Construct Array Of Controllers, In Our Case Just The One
-	controllers := [...]*kncontroller.Impl{
-		controller.NewController(
-			logger,
-			environment.ChannelKey,
-			dispatcher,
-			kafkaChannelInformer,
-			k8sClient,
-			kafkaClient,
-			ctx.Done(),
-			managerEvents,
-		),
+	// Construct The KafkaChannel Controller
+	kcController := controller.NewController(
+		logger,
+		environment.ChannelKey,
+		dispatcher,
+		kafkaChannelInformer,
+		k8sClient,
+		kafkaClient,
+		ctx.Done(),
+		managerEvents,
+	)
+
+	// Watch The Secret For Changes
+	secretObserver := NewSecretObserver(kcController, environment.ChannelKey, dispatcher)
+	err = distributedcommonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, environment.ResyncPeriod, secretObserver)
+	if err != nil {
+		logger.Fatal("Failed To Start Secret Watcher", zap.Error(err))
 	}
 
 	// Start The Informers
-	logger.Info("Starting informers.")
+	logger.Info("Starting Informers")
 	if err := kncontroller.StartInformers(ctx.Done(), kafkaChannelInformer.Informer()); err != nil {
 		logger.Error("Failed to start informers", zap.Error(err))
 		return
@@ -201,8 +192,8 @@ func main() {
 	healthServer.SetDispatcherReady(true)
 
 	// Start The Controllers (Blocking WaitGroup.Wait Call)
-	logger.Info("Starting controllers.")
-	kncontroller.StartAll(ctx, controllers[:]...)
+	logger.Info("Starting KafkaChannel Controller")
+	kncontroller.StartAll(ctx, kcController)
 
 	// Reset The Liveness and Readiness Flags In Preparation For Shutdown
 	healthServer.Shutdown()
@@ -219,21 +210,35 @@ func flush(logger *zap.Logger) {
 	eventingmetrics.FlushExporter()
 }
 
-// secretObserver is the callback function that handles changes to our Secret
-func secretObserver(ctx context.Context, secret *corev1.Secret) {
-	secretLogger := logging.FromContext(ctx)
-
-	if secret == nil {
-		secretLogger.Warn("Nil Secret passed to secretObserver; ignoring")
-		return
+// NewSecretObserver is a factory for creating the callback function that handles changes to the Kafka Secret.
+func NewSecretObserver(kcController *kncontroller.Impl, channelKey string, dispatcher dispatch.Dispatcher) func(ctx context.Context, secret *corev1.Secret) {
+	return func(ctx context.Context, secret *corev1.Secret) {
+
+		// Get The Logger From The Context
+		logger := logging.FromContext(ctx)
+
+		// Validate The Secret (Ignore Invalid)
+		if secret == nil {
+			logger.Warn("Nil Secret passed to secretObserver; ignoring")
+			return
+		}
+
+		// Ignore Startup Scenario Where Dispatcher Reference Might Be Nil
+		if dispatcher == nil {
+			logger.Debug("Dispatcher is nil during call to secretObserver; ignoring changes")
+			return
+		}
+
+		// Signal The Dispatcher To Recreate Kafka Config And Reconnect All Current ConsumerGroups
+		dispatcher.SecretChanged(ctx, secret)
+
+		// Requeue The KafkaChannel To Fix Any Not-Ready Subscriptions
+		channelNamespace, channelName, err := cache.SplitMetaNamespaceKey(channelKey)
+		if err != nil {
+			logger.Error("Invalid KafkaChannel Key", zap.String("ChannelKey", channelKey), zap.Error(err))
+		} else {
+			logger.Debug("Requeue-ing KafkaChannel due to Kafka Secret change", zap.String("ChannelKey", channelKey))
+			kcController.EnqueueKey(types.NamespacedName{Namespace: channelNamespace, Name: channelName})
+		}
 	}
-
-	if dispatcher == nil {
-		// This typically happens during startup
-		secretLogger.Debug("Dispatcher is nil during call to secretObserver; ignoring changes")
-		return
-	}
-
-	// Toss the new secret to the dispatcher for inspection and action
-	dispatcher.SecretChanged(ctx, secret)
 }
diff --git a/cmd/channel/distributed/receiver/main.go b/cmd/channel/distributed/receiver/main.go
index da826ad05..65f7063ee 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -52,10 +52,7 @@ import (
 )
 
 // Variables
-var (
-	logger        *zap.Logger
-	kafkaProducer *producer.Producer
-)
+var kafkaProducer *producer.Producer
 
 // The Main Function (Go Command)
 func main() {
@@ -78,7 +75,7 @@ func main() {
 	ctx = commonk8s.LoggingContext(ctx, constants.Component, k8sClient)
 
 	// Get The Logger From The Context & Defer Flushing Any Buffered Log Entries On Exit
-	logger = logging.FromContext(ctx).Desugar()
+	logger := logging.FromContext(ctx).Desugar()
 	defer flush(logger)
 
 	// Load Environment Variables
@@ -155,7 +152,7 @@ func main() {
 	if err != nil {
 		logger.Fatal("Failed To Initialize Kafka Producer", zap.Error(err))
 	}
-	defer kafkaProducer.Close()
+	defer closeProducer()
 
 	channelReporter := eventingchannel.NewStatsReporter(environment.ContainerName, kmeta.ChildName(environment.PodName, uuid.New().String()))
 
@@ -187,6 +184,9 @@ func flush(logger *zap.Logger) {
 // CloudEvent Message Handler - Converts To KafkaMessage And Produces To Channel's Kafka Topic
 func handleMessage(ctx context.Context, channelReference eventingchannel.ChannelReference, message binding.Message, transformers []binding.Transformer, _ http.Header) error {
 
+	// Get The Logger From The Context
+	logger := logging.FromContext(ctx).Desugar()
+
 	// Note - The context provided here is a different context from the one created in main() and does not have our logger instance.
 	if logger.Core().Enabled(zap.DebugLevel) {
 		// Checked Logging Level First To Avoid Calling zap.Any In Production
@@ -215,17 +215,27 @@ func handleMessage(ctx context.Context, channelReference eventingchannel.Channel
 	return nil
 }
 
+// closeProducer performs a safe close on the current "global" kafkaProducer instance.
+func closeProducer() {
+	if kafkaProducer != nil {
+		kafkaProducer.Close()
+	}
+}
+
 // secretObserver is the callback function that handles changes to our Secret
 func secretObserver(ctx context.Context, secret *corev1.Secret) {
+
+	// Get The Logger From The Context
 	logger := logging.FromContext(ctx)
 
+	// Validate The Secret (Ignore Invalid)
 	if secret == nil {
 		logger.Warn("Nil Secret passed to secretObserver; ignoring")
 		return
 	}
 
+	// Ignore Startup Scenario Where Producer Reference Might Be Nil
 	if kafkaProducer == nil {
-		// This typically happens during startup
 		logger.Debug("Producer is nil during call to secretObserver; ignoring changes")
 		return
 	}
diff --git a/go.mod b/go.mod
index d8dfc1c2b..de584bdaf 100644
--- a/go.mod
+++ b/go.mod
@@ -35,7 +35,7 @@ require (
 	k8s.io/client-go v0.20.7
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
 	knative.dev/control-protocol v0.0.0-20210803191615-72cde96b9f5a
-	knative.dev/eventing v0.24.1-0.20210803173815-5c7b55336efd
+	knative.dev/eventing v0.24.1-0.20210805140952-ca9e469d46db
 	knative.dev/hack v0.0.0-20210622141627-e28525d8d260
 	knative.dev/networking v0.0.0-20210803181815-acdfd41c575c
 	knative.dev/pkg v0.0.0-20210803160015-21eb4c167cc5
diff --git a/go.sum b/go.sum
index 8baa63cf1..9c177d638 100644
--- a/go.sum
+++ b/go.sum
@@ -1282,8 +1282,8 @@ k8s.io/utils v0.0.0-20201110183641-67b214c5f920 h1:CbnUZsM497iRC5QMVkHwyl8s2tB3g
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20210803191615-72cde96b9f5a h1:VbJj20YhSx3j+OE8DhncTt+TYQrb+rAS1BGPoeo3GF0=
 knative.dev/control-protocol v0.0.0-20210803191615-72cde96b9f5a/go.mod h1:9gYmAq6zVKUA+p1+DXslH91jtga5q2z+YJWcH+z88EE=
-knative.dev/eventing v0.24.1-0.20210803173815-5c7b55336efd h1:ygV2agIpofSLAp34xT5MDYFBhoEHUVCIO8O2BtdH35g=
-knative.dev/eventing v0.24.1-0.20210803173815-5c7b55336efd/go.mod h1:mYN9AeU95zB9QT9ATmYiq0i7k41u69xD0jW1udHdPrc=
+knative.dev/eventing v0.24.1-0.20210805140952-ca9e469d46db h1:C2LRDcj3QorrxIbrY4BbHomXjYSsNJYHjKNcEougmyA=
+knative.dev/eventing v0.24.1-0.20210805140952-ca9e469d46db/go.mod h1:mYN9AeU95zB9QT9ATmYiq0i7k41u69xD0jW1udHdPrc=
 knative.dev/hack v0.0.0-20210622141627-e28525d8d260 h1:f2eMtOubAOc/Q7JlvFPDKXiPlJVK+VpX2Cot8hRzCgQ=
 knative.dev/hack v0.0.0-20210622141627-e28525d8d260/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210622141627-e28525d8d260/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
diff --git a/pkg/channel/distributed/common/kafka/util/util.go b/pkg/channel/distributed/common/kafka/util/util.go
index 06497aad4..658586b25 100644
--- a/pkg/channel/distributed/common/kafka/util/util.go
+++ b/pkg/channel/distributed/common/kafka/util/util.go
@@ -20,9 +20,13 @@ import (
 	"fmt"
 	"strings"
 
+	"k8s.io/apimachinery/pkg/types"
+
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
 )
 
+const GroupIdPrefix = "kafka"
+
 // TopicName returns a formatted string representing the Kafka Topic name.
 func TopicName(namespace string, name string) string {
 	return fmt.Sprintf("%s.%s", namespace, name)
@@ -30,7 +34,12 @@ func TopicName(namespace string, name string) string {
 
 // GroupId returns a formatted string representing the Kafka ConsumerGroup ID.
 func GroupId(uid string) string {
-	return fmt.Sprintf("kafka.%s", uid)
+	return fmt.Sprintf("%s.%s", GroupIdPrefix, uid)
+}
+
+// Uid returns a UID from the specified GroupId, which is the inverse of GroupId().
+func Uid(groupId string) types.UID {
+	return types.UID(strings.TrimPrefix(groupId, GroupIdPrefix+"."))
 }
 
 // AppendKafkaChannelServiceNameSuffix appends the KafkaChannel Service name suffix to the specified string.
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
index a319de7e1..c8be54863 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
@@ -90,7 +90,10 @@ func NewController(
 		kafkachannelLister:   kafkachannelInformer.Lister(),
 		kafkaClientSet:       kafkaClientSet,
 	}
-	reconciler.impl = controller.NewImpl(reconciler, reconciler.logger.Sugar(), ReconcilerName)
+	reconciler.impl = controller.NewImplFull(reconciler, controller.ControllerOptions{
+		WorkQueueName: ReconcilerName,
+		Logger:        logger.Sugar(),
+	})
 
 	reconciler.logger.Info("Setting Up Event Handlers")
 
@@ -118,7 +121,7 @@ func NewController(
 }
 
 // processManagerEvents will listen on the channel provided by the KafkaConsumerGroupManager for events
-// related to changes in the status of a managed ConsumerGroup.  This function is non-blocking, and the
+// related to a change in the status of a managed ConsumerGroup.  This function is non-blocking, and the
 // internal goroutine will exit when the channel is closed.
 func (r Reconciler) processManagerEvents(events <-chan commonconsumer.ManagerEvent) error {
 	// If the events channel is nil, there's no point listening to it; it will just block forever
@@ -168,7 +171,7 @@ func (r Reconciler) Reconcile(ctx context.Context, key string) error {
 	// Convert the namespace/name string into a distinct namespace and name.
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
-		logging.FromContext(ctx).Error("invalid resource key")
+		logging.FromContext(ctx).Error("invalid resource key", zap.String("Key", key), zap.Error(err))
 		return nil
 	}
 
@@ -190,6 +193,7 @@ func (r Reconciler) Reconcile(ctx context.Context, key string) error {
 	// Don't modify the informers copy
 	channel := original.DeepCopy()
 
+	// Perform the reconciliation (will update KafkaChannel.Status)
 	reconcileError := r.reconcile(channel)
 	if reconcileError != nil {
 		r.logger.Error("Error Reconciling KafkaChannel", zap.Error(reconcileError))
@@ -199,6 +203,7 @@ func (r Reconciler) Reconcile(ctx context.Context, key string) error {
 		r.recorder.Event(channel, corev1.EventTypeNormal, channelReconciled, "KafkaChannel Reconciled")
 	}
 
+	// Push KafkaChannel Status Changes To K8S
 	_, updateStatusErr := r.updateStatus(ctx, channel)
 	if updateStatusErr != nil {
 		r.logger.Error("Failed To Update KafkaChannel Status", zap.Error(updateStatusErr))
@@ -208,8 +213,8 @@ func (r Reconciler) Reconcile(ctx context.Context, key string) error {
 		r.logger.Info("Successfully Verified / Updated KafkaChannel Status")
 	}
 
-	// Return Success
-	return nil
+	// Return Reconciliation Errors To Requeue
+	return reconcileError
 }
 
 // Reconcile The Specified KafkaChannel
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index a10ac2d11..293602eb2 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -48,7 +48,6 @@ type DispatcherConfig struct {
 	StatsReporter   metrics.StatsReporter
 	MetricsRegistry gometrics.Registry
 	SaramaConfig    *sarama.Config
-	SubscriberSpecs []eventingduck.SubscriberSpec
 }
 
 // SubscriberWrapper Defines A Knative Eventing SubscriberSpec Wrapper Enhanced With Sarama ConsumerGroup ID
@@ -138,7 +137,7 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 	d.consumerUpdateLock.Lock()
 	defer d.consumerUpdateLock.Unlock()
 
-	// Loop Over All All The Specified Subscribers
+	// Loop Over All The Specified Subscribers
 	for _, subscriberSpec := range subscriberSpecs {
 
 		// Format The GroupId For The Specified Subscriber
@@ -158,6 +157,7 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 				// Log & Return Failure
 				logger.Error("Failed To Create ConsumerGroup", zap.Error(err))
 				subscriptions[subscriberSpec.UID] = commonconsumer.SubscriberStatus{Error: err}
+
 			} else {
 
 				// Create A New SubscriberWrapper With The ConsumerGroup
@@ -178,7 +178,8 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 			}
 
 		} else {
-			// Otherwise Just Add To List Of Active Subscribers
+
+			// Otherwise, Just Add To List Of Active Subscribers
 			subscriptions[subscriberSpec.UID] = commonconsumer.SubscriberStatus{}
 
 			// If the group is stopped, it's still active but the reconciler needs to know about it in order
@@ -190,17 +191,11 @@ func (d *DispatcherImpl) UpdateSubscriptions(subscriberSpecs []eventingduck.Subs
 		}
 	}
 
-	// Save the current (active) subscriber specs so that SecretChanged() can use them to recreate the Dispatcher
-	// if necessary without going through the inactive subscribers again.
-	d.SubscriberSpecs = []eventingduck.SubscriberSpec{}
-
 	// Close ConsumerGroups For Removed/Failed Subscriptions (In Map But No Longer Active)
 	for _, subscriber := range d.subscribers {
 		subscription, ok := subscriptions[subscriber.UID]
 		if !ok || subscription.Error != nil {
 			d.closeConsumerGroup(subscriber)
-		} else {
-			d.SubscriberSpecs = append(d.SubscriberSpecs, subscriber.SubscriberSpec)
 		}
 	}
 
@@ -273,10 +268,16 @@ func (d *DispatcherImpl) SecretChanged(ctx context.Context, secret *corev1.Secre
 	d.DispatcherConfig.SaramaConfig = newConfig
 
 	// Replace The Dispatcher's ConsumerGroupFactory With Updated Version Using New Config
-	// Note:  This will close and recreate all of the managed consumer groups
-	err = d.consumerMgr.Reconfigure(d.DispatcherConfig.Brokers, d.DispatcherConfig.SaramaConfig)
-	if err != nil {
-		d.Logger.Error("Failed To Reconfigure Consumer Group Manager Using Updated Secret", zap.Error(err))
+	// Note:  This will close and recreate all managed ConsumerGroups
+	reconfigureErr := d.consumerMgr.Reconfigure(d.DispatcherConfig.Brokers, d.DispatcherConfig.SaramaConfig)
+	if reconfigureErr != nil {
+
+		// Remove All Failed Subscribers From List To Allow Recreation Next Reconcile Loop (Expects Caller To Requeue KafkaChannel!)
+		d.Logger.Error("Failed To Reconfigure Consumer Group Manager Using Updated Secret", zap.Error(reconfigureErr))
+		for _, groupId := range reconfigureErr.GroupIds {
+			subscriberUID := commonkafkautil.Uid(groupId)
+			delete(d.subscribers, subscriberUID)
+		}
 	}
 }
 
diff --git a/pkg/common/consumer/consumer_manager.go b/pkg/common/consumer/consumer_manager.go
index 8279b31b8..ff6b24bfd 100644
--- a/pkg/common/consumer/consumer_manager.go
+++ b/pkg/common/consumer/consumer_manager.go
@@ -92,7 +92,7 @@ type ManagerEvent struct {
 
 // KafkaConsumerGroupManager keeps track of Sarama consumer groups and handles messages from control-protocol clients
 type KafkaConsumerGroupManager interface {
-	Reconfigure(brokers []string, config *sarama.Config) error
+	Reconfigure(brokers []string, config *sarama.Config) *ReconfigureError
 	StartConsumerGroup(groupId string, topics []string, logger *zap.SugaredLogger, handler KafkaConsumerHandler, options ...SaramaConsumerHandlerOption) error
 	CloseConsumerGroup(groupId string) error
 	Errors(groupId string) <-chan error
@@ -120,6 +120,24 @@ type kafkaConsumerGroupManagerImpl struct {
 // Verify that the kafkaConsumerGroupManagerImpl satisfies the KafkaConsumerGroupManager interface
 var _ KafkaConsumerGroupManager = (*kafkaConsumerGroupManagerImpl)(nil)
 
+// ReconfigureError is a custom error type returned by the Reconfigure() function.
+type ReconfigureError struct {
+	MultiError error    // A MultiError with all ConsumerGroup failures.
+	GroupIds   []string // The ConsumerGroup IDs which failed to stop/start.
+}
+
+// Error implements the golang error interface.
+func (r ReconfigureError) Error() string {
+	var multiErrString string
+	if r.MultiError != nil {
+		multiErrString = r.MultiError.Error()
+	}
+	return fmt.Sprintf("Reconfigure Failed: MultiErr='%s', GroupIds='%+v'", multiErrString, r.GroupIds)
+}
+
+// Ensure ReconfigureError Is A Valid Error
+var _ error = (*ReconfigureError)(nil)
+
 // NewConsumerGroupManager returns a new kafkaConsumerGroupManagerImpl as a KafkaConsumerGroupManager interface
 func NewConsumerGroupManager(logger *zap.Logger, serverHandler controlprotocol.ServerHandler, brokers []string, config *sarama.Config) KafkaConsumerGroupManager {
 
@@ -189,21 +207,32 @@ func (m *kafkaConsumerGroupManagerImpl) notify(event ManagerEvent) {
 
 // Reconfigure will incorporate a new set of brokers and Sarama config settings into the manager
 // without requiring a new control-protocol server or losing the current map of managed groups.
-// It will stop and start all of the managed groups in the group map.
-func (m *kafkaConsumerGroupManagerImpl) Reconfigure(brokers []string, config *sarama.Config) error {
+// It will stop and start all the managed groups in the group map.
+func (m *kafkaConsumerGroupManagerImpl) Reconfigure(brokers []string, config *sarama.Config) *ReconfigureError {
+
 	m.logger.Info("Reconfigure Consumer Group Manager - Stopping All Managed Consumer Groups")
+
 	var multiErr error
+	var failedGroups []string
 	groupsToRestart := make([]string, 0, len(m.groups))
 	for groupId := range m.groups {
-		err := m.stopConsumerGroup(&commands.CommandLock{Token: internalToken, LockBefore: true}, groupId)
-		if err != nil {
-			// If we couldn't stop a group, or failed to obtain a lock, note it as an error.  However,
-			// in a practical sense, the new brokers/config will be used when whatever locked the group
-			// restarts it anyway.
-			multierr.AppendInto(&multiErr, err)
-		} else {
-			// Only attempt to restart groups that this function stopped
-			groupsToRestart = append(groupsToRestart, groupId)
+
+		// Do NOT Attempt To Reconfigure Stopped ConsumerGroups! (Would unintentionally start previously stopped ConsumerGroups)
+		if !m.IsStopped(groupId) {
+			err := m.stopConsumerGroup(&commands.CommandLock{Token: internalToken, LockBefore: true}, groupId)
+			if err != nil {
+				// If we couldn't stop a group, or failed to obtain a lock, note it as an error.  However,
+				// in a practical sense, the new brokers/config will be used when whatever locked the group
+				// restarts it anyway.
+				multierr.AppendInto(&multiErr, err)
+
+				// Remove The Group From The Manager So That It Will Be Restarted On Next Reconciliation
+				m.removeGroup(groupId)
+				failedGroups = append(failedGroups, groupId)
+			} else {
+				// Only attempt to restart groups that this function stopped
+				groupsToRestart = append(groupsToRestart, groupId)
+			}
 		}
 	}
 
@@ -215,9 +244,19 @@ func (m *kafkaConsumerGroupManagerImpl) Reconfigure(brokers []string, config *sa
 		err := m.startConsumerGroup(&commands.CommandLock{Token: internalToken, UnlockAfter: true}, groupId)
 		if err != nil {
 			multierr.AppendInto(&multiErr, err)
+
+			// Remove The Group From The Manager So That It Will Be Restarted On Next Reconciliation
+			m.removeGroup(groupId)
+			failedGroups = append(failedGroups, groupId)
 		}
 	}
-	return multiErr
+
+	// Create & Return ReconfigureError If Necessary
+	if multiErr != nil {
+		return &ReconfigureError{MultiError: multiErr, GroupIds: failedGroups}
+	} else {
+		return nil
+	}
 }
 
 // StartConsumerGroup uses the consumer factory to create a new ConsumerGroup, add it to the list
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 1ff996641..b92431b0c 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1046,7 +1046,7 @@ knative.dev/control-protocol/pkg/message
 knative.dev/control-protocol/pkg/network
 knative.dev/control-protocol/pkg/reconciler
 knative.dev/control-protocol/pkg/service
-# knative.dev/eventing v0.24.1-0.20210803173815-5c7b55336efd
+# knative.dev/eventing v0.24.1-0.20210805140952-ca9e469d46db
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test


Test patch:
diff --git a/pkg/channel/distributed/common/kafka/util/util_test.go b/pkg/channel/distributed/common/kafka/util/util_test.go
index 7d3fa0449..d7b5fa375 100644
--- a/pkg/channel/distributed/common/kafka/util/util_test.go
+++ b/pkg/channel/distributed/common/kafka/util/util_test.go
@@ -21,6 +21,8 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/types"
+
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
 )
 
@@ -53,6 +55,21 @@ func TestGroupId(t *testing.T) {
 	assert.Equal(t, expectedGroupId, actualGroupId)
 }
 
+// Test The Uid() Functionality
+func TestUid(t *testing.T) {
+
+	// Test Data
+	uidString := "TestUID"
+	groupId := GroupIdPrefix + "." + uidString
+
+	// Perform The Test
+	actualUid := Uid(groupId)
+
+	// Verify The Results
+	expectedUid := types.UID(uidString)
+	assert.Equal(t, expectedUid, actualUid)
+}
+
 // Test The AppendChannelServiceNameSuffix() Functionality
 func TestAppendChannelServiceNameSuffix(t *testing.T) {
 
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
index 8efe528ec..64c37bc65 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
@@ -290,7 +290,7 @@ func TestAllCases(t *testing.T) {
 					reconciletesting.WithSubscriberReady("1")),
 			},
 			Key:     kcKey,
-			WantErr: false,
+			WantErr: true,
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 				Object: reconciletesting.NewKafkaChannel(kcName, testNS,
 					reconciletesting.WithInitKafkaChannelConditions,
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
index 467e4126d..d0cf85967 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
@@ -32,9 +32,11 @@ import (
 	"knative.dev/pkg/logging"
 	logtesting "knative.dev/pkg/logging/testing"
 
+	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/util"
 	commonclient "knative.dev/eventing-kafka/pkg/common/client"
 	clienttesting "knative.dev/eventing-kafka/pkg/common/client/testing"
 	configtesting "knative.dev/eventing-kafka/pkg/common/config/testing"
+	"knative.dev/eventing-kafka/pkg/common/consumer"
 	consumertesting "knative.dev/eventing-kafka/pkg/common/consumer/testing"
 	controltesting "knative.dev/eventing-kafka/pkg/common/controlprotocol/testing"
 	kafkatesting "knative.dev/eventing-kafka/pkg/common/kafka/testing"
@@ -508,7 +510,10 @@ func TestSecretChanged(t *testing.T) {
 			impl := dispatcher.(*DispatcherImpl)
 			impl.subscribers = map[types.UID]*SubscriberWrapper{uid123: createSubscriberWrapper(uid123)}
 			if testCase.reconfigureErr {
-				mockManager.On("Reconfigure", mock.Anything, mock.Anything).Return(fmt.Errorf("reconfigure error"))
+				mockManager.On("Reconfigure", mock.Anything, mock.Anything).Return(&consumer.ReconfigureError{
+					MultiError: fmt.Errorf("reconfigure error"),
+					GroupIds:   []string{util.GroupId(string(uid123))},
+				})
 				impl.consumerMgr = mockManager
 			}
 
@@ -530,6 +535,9 @@ func TestSecretChanged(t *testing.T) {
 			if testCase.expectNewSaslType != "" {
 				assert.Equal(t, testCase.expectNewSaslType, string(impl.SaramaConfig.Net.SASL.Mechanism))
 			}
+			if testCase.reconfigureErr {
+				assert.Nil(t, impl.subscribers[uid123])
+			}
 
 			mockManager.AssertExpectations(t)
 		})
@@ -550,9 +558,6 @@ func createTestDispatcher(t *testing.T, brokers []string, config *sarama.Config)
 	// Create StatsReporter
 	statsReporter := metrics.NewStatsReporter(logger)
 
-	// Create An Empty Set Of SubscriberSpecs
-	subscriberSpecs := make([]eventingduck.SubscriberSpec, 0)
-
 	// Create The DispatcherConfig
 	dispatcherConfig := DispatcherConfig{
 		Logger:          logger,
@@ -560,7 +565,6 @@ func createTestDispatcher(t *testing.T, brokers []string, config *sarama.Config)
 		StatsReporter:   statsReporter,
 		MetricsRegistry: config.MetricRegistry,
 		SaramaConfig:    config,
-		SubscriberSpecs: subscriberSpecs,
 	}
 
 	serverHandler := controltesting.GetMockServerHandler()
diff --git a/pkg/common/consumer/consumer_manager_test.go b/pkg/common/consumer/consumer_manager_test.go
index 86b75fdbc..2c4c86b34 100644
--- a/pkg/common/consumer/consumer_manager_test.go
+++ b/pkg/common/consumer/consumer_manager_test.go
@@ -66,13 +66,13 @@ func TestReconfigure(t *testing.T) {
 			name:      "Error stopping groups",
 			groupId:   "test-id1",
 			closeErr:  fmt.Errorf("error closing consumer group"),
-			expectErr: "error closing consumer group",
+			expectErr: "Reconfigure Failed: MultiErr='error closing consumer group', GroupIds='[test-id1]'",
 		},
 		{
 			name:       "Error starting groups",
 			groupId:    "test-id1",
 			factoryErr: true,
-			expectErr:  "factory error",
+			expectErr:  "Reconfigure Failed: MultiErr='factory error', GroupIds='[test-id1]'",
 		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
@@ -86,12 +86,59 @@ func TestReconfigure(t *testing.T) {
 			} else {
 				assert.NotNil(t, err)
 				assert.Equal(t, testCase.expectErr, err.Error())
+				assert.Len(t, err.GroupIds, 1)
+				assert.Equal(t, testCase.groupId, err.GroupIds[0])
 			}
 			server.AssertExpectations(t)
 		})
 	}
 }
 
+func TestReconfigureError(t *testing.T) {
+	for _, testCase := range []struct {
+		name   string
+		err    ReconfigureError
+		result string
+	}{
+		{
+			name: "Empty",
+			err: ReconfigureError{
+				MultiError: nil,
+				GroupIds:   nil,
+			},
+			result: "Reconfigure Failed: MultiErr='', GroupIds='[]'",
+		},
+		{
+			name: "MultiErr Only",
+			err: ReconfigureError{
+				MultiError: fmt.Errorf("test-error"),
+				GroupIds:   nil,
+			},
+			result: "Reconfigure Failed: MultiErr='test-error', GroupIds='[]'",
+		},
+		{
+			name: "GroupIds Only",
+			err: ReconfigureError{
+				MultiError: nil,
+				GroupIds:   []string{"one", "two", "three"},
+			},
+			result: "Reconfigure Failed: MultiErr='', GroupIds='[one two three]'",
+		},
+		{
+			name: "Complete",
+			err: ReconfigureError{
+				MultiError: fmt.Errorf("test-error"),
+				GroupIds:   []string{"one", "two", "three"},
+			},
+			result: "Reconfigure Failed: MultiErr='test-error', GroupIds='[one two three]'",
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			assert.Equal(t, testCase.result, testCase.err.Error())
+		})
+	}
+}
+
 func TestStartConsumerGroup(t *testing.T) {
 	defer restoreNewConsumerGroup(newConsumerGroup)
 
diff --git a/pkg/common/consumer/testing/mocks.go b/pkg/common/consumer/testing/mocks.go
index 73e6bd29a..9a7f40bac 100644
--- a/pkg/common/consumer/testing/mocks.go
+++ b/pkg/common/consumer/testing/mocks.go
@@ -56,8 +56,8 @@ func NewMockConsumerGroupManager() *MockConsumerGroupManager {
 
 var _ consumer.KafkaConsumerGroupManager = (*MockConsumerGroupManager)(nil)
 
-func (m *MockConsumerGroupManager) Reconfigure(brokers []string, config *sarama.Config) error {
-	return m.Called(brokers, config).Error(0)
+func (m *MockConsumerGroupManager) Reconfigure(brokers []string, config *sarama.Config) *consumer.ReconfigureError {
+	return m.Called(brokers, config).Get(0).(*consumer.ReconfigureError)
 }
 
 func (m *MockConsumerGroupManager) StartConsumerGroup(groupId string, topics []string, logger *zap.SugaredLogger,
diff --git a/test/rekt/features/kafkasource/data_plane.go b/test/rekt/features/kafkasource/data_plane.go
index 6f48ba729..5f9a24d5b 100644
--- a/test/rekt/features/kafkasource/data_plane.go
+++ b/test/rekt/features/kafkasource/data_plane.go
@@ -28,6 +28,7 @@ import (
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/eventshub/assert"
 	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
 	"knative.dev/reconciler-test/pkg/state"
 
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
@@ -259,7 +260,7 @@ func dataPlaneDelivery(name string, data delivery) *feature.Feature {
 	// Setup source
 
 	// options
-	ksopts := []kafkasource.CfgFn{
+	ksopts := []manifest.CfgFn{
 		kafkasource.WithBootstrapServers([]string{data.bootstrapServer}),
 		kafkasource.WithTopics([]string{topicName}),
 		kafkasource.WithSink(&duckv1.KReference{
@@ -306,7 +307,7 @@ func sinkReceiveEvent(name string, matchers EventMatcher) func(ctx context.Conte
 		sinkName := state.GetStringOrFail(ctx, t, "sinkName")
 
 		// Install kafkacat
-		kcopts := []kafkacat.CfgFn{
+		kcopts := []manifest.CfgFn{
 			kafkacat.WithBootstrapServer(kafkaBootstrapUrlPlain),
 			kafkacat.WithTopic(topicName),
 			kafkacat.WithKey(message.Key),
diff --git a/test/rekt/features/kafkasource/mt_data_plane.go b/test/rekt/features/kafkasource/mt_data_plane.go
index bdee8ac45..da82e3a72 100644
--- a/test/rekt/features/kafkasource/mt_data_plane.go
+++ b/test/rekt/features/kafkasource/mt_data_plane.go
@@ -30,6 +30,7 @@ import (
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/eventshub/assert"
 	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
 	"knative.dev/reconciler-test/pkg/state"
 
 	kafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client"
@@ -79,7 +80,7 @@ func MtDataPlaneDelivery() *feature.Feature {
 
 	f.Setup("install sink", eventshub.Install(sinkName, eventshub.StartReceiver))
 
-	ksopts := []kafkasource.CfgFn{
+	ksopts := []manifest.CfgFn{
 		kafkasource.WithBootstrapServers([]string{delivery.bootstrapServer}),
 		kafkasource.WithTopics([]string{topicName}),
 		kafkasource.WithSink(&duckv1.KReference{
@@ -109,7 +110,7 @@ func sourcesPlaced(name string, matchers EventMatcher) func(ctx context.Context,
 			time.Sleep(10 * time.Second)
 		}
 
-		kcopts := []kafkacat.CfgFn{
+		kcopts := []manifest.CfgFn{
 			kafkacat.WithBootstrapServer(kafkaBootstrapUrlPlain),
 			kafkacat.WithTopic(topicName),
 			kafkacat.WithKey(message.Key),
diff --git a/test/rekt/features/kafkasource/readiness.go b/test/rekt/features/kafkasource/readiness.go
index 49066f910..dc0bcfa97 100644
--- a/test/rekt/features/kafkasource/readiness.go
+++ b/test/rekt/features/kafkasource/readiness.go
@@ -20,13 +20,14 @@ import (
 	"knative.dev/eventing/test/rekt/resources/svc"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
 
 	"knative.dev/eventing-kafka/test/rekt/resources/kafkasource"
 	"knative.dev/eventing-kafka/test/rekt/resources/kafkatopic"
 )
 
 // KafkaSourceGoesReady returns a feature testing if a KafkaSource becomes ready.
-func KafkaSourceGoesReady(name string, cfg ...kafkasource.CfgFn) *feature.Feature {
+func KafkaSourceGoesReady(name string, cfg ...manifest.CfgFn) *feature.Feature {
 	f := feature.NewFeatureNamed("KafkaSource goes ready.")
 
 	topicName := feature.MakeRandomK8sName("topic") // A k8s name is also a valid topic name.
diff --git a/test/rekt/features/kafkatopic/readiness.go b/test/rekt/features/kafkatopic/readiness.go
index a57a28648..576c973f8 100644
--- a/test/rekt/features/kafkatopic/readiness.go
+++ b/test/rekt/features/kafkatopic/readiness.go
@@ -18,12 +18,13 @@ package kafkatopic
 
 import (
 	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/manifest"
 
 	"knative.dev/eventing-kafka/test/rekt/resources/kafkatopic"
 )
 
 // KafkaTopicGoesReady returns a feature testing if a Kafka topic becomes ready.
-func KafkaTopicGoesReady(name string, cfg ...kafkatopic.CfgFn) *feature.Feature {
+func KafkaTopicGoesReady(name string, cfg ...manifest.CfgFn) *feature.Feature {
 	f := feature.NewFeatureNamed("KafkaTopic goes ready.")
 
 	f.Setup("install a Kafka topic", kafkatopic.Install(name, cfg...))
diff --git a/test/rekt/resources/kafkacat/kafkacat.go b/test/rekt/resources/kafkacat/kafkacat.go
index f7d455789..67fddbfd2 100644
--- a/test/rekt/resources/kafkacat/kafkacat.go
+++ b/test/rekt/resources/kafkacat/kafkacat.go
@@ -18,16 +18,18 @@ package kafkacat
 
 import (
 	"context"
+	"embed"
 
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/manifest"
 )
 
-type CfgFn func(map[string]interface{})
+//go:embed *.yaml
+var yaml embed.FS
 
 // Install will create a pod running kafkacat, augmented with the config fn options.
-func Install(name string, opts ...CfgFn) feature.StepFn {
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 	cfg := map[string]interface{}{
 		"name": name,
 	}
@@ -36,8 +38,8 @@ func Install(name string, opts ...CfgFn) feature.StepFn {
 	}
 
 	return func(ctx context.Context, t feature.T) {
-		if _, err := manifest.InstallLocalYaml(ctx, cfg); err != nil {
-			t.Fatal(err, cfg)
+		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+			t.Fatal(err)
 		}
 
 		k8s.WaitForPodRunningOrFail(ctx, t, name)
@@ -45,28 +47,28 @@ func Install(name string, opts ...CfgFn) feature.StepFn {
 }
 
 // WithPayload adds payload to the kafkacat configmap.
-func WithPayload(payload string) CfgFn {
+func WithPayload(payload string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		cfg["payload"] = payload
 	}
 }
 
 // WithBootstrapServer adds bootstrapServer to the kafkacat argument list.
-func WithBootstrapServer(bootstrapServer string) CfgFn {
+func WithBootstrapServer(bootstrapServer string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		cfg["bootstrapServer"] = bootstrapServer
 	}
 }
 
 // WithTopic adds the topic to the kafkacat argument list.
-func WithTopic(topic string) CfgFn {
+func WithTopic(topic string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		cfg["topic"] = topic
 	}
 }
 
 // WithKey adds the key to the kafkacat argument list.
-func WithKey(key string) CfgFn {
+func WithKey(key string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if key != "" {
 			cfg["key"] = key
@@ -75,7 +77,7 @@ func WithKey(key string) CfgFn {
 }
 
 // WithHeaders adds the headers to the kafkacat argument list.
-func WithHeaders(headers map[string]string) CfgFn {
+func WithHeaders(headers map[string]string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if len(headers) > 0 {
 			cfg["headers"] = headers
diff --git a/test/rekt/resources/kafkacat/kafkacat_test.go b/test/rekt/resources/kafkacat/kafkacat_test.go
index e80d68ff8..20b12bfc1 100644
--- a/test/rekt/resources/kafkacat/kafkacat_test.go
+++ b/test/rekt/resources/kafkacat/kafkacat_test.go
@@ -29,7 +29,7 @@ func Example_min() {
 		"namespace": "bar",
 	}
 
-	files, err := manifest.ExecuteLocalYAML(images, cfg)
+	files, err := manifest.ExecuteYAML(yaml, images, cfg)
 	if err != nil {
 		panic(err)
 	}
diff --git a/test/rekt/resources/kafkachannel/kafkachannel.go b/test/rekt/resources/kafkachannel/kafkachannel.go
index 3c4007886..9408eae1e 100644
--- a/test/rekt/resources/kafkachannel/kafkachannel.go
+++ b/test/rekt/resources/kafkachannel/kafkachannel.go
@@ -18,6 +18,7 @@ package kafkachannel
 
 import (
 	"context"
+	"embed"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -26,14 +27,15 @@ import (
 	"knative.dev/reconciler-test/pkg/manifest"
 )
 
-type CfgFn func(map[string]interface{})
+//go:embed *.yaml
+var yaml embed.FS
 
 func GVR() schema.GroupVersionResource {
 	return schema.GroupVersionResource{Group: "messaging.knative.dev", Version: "v1beta1", Resource: "kafkachannels"}
 }
 
 // Install will create a KafkaChannel resource, using the latest version, augmented with the config fn options.
-func Install(name string, opts ...CfgFn) feature.StepFn {
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 	cfg := map[string]interface{}{
 		"name":    name,
 		"version": GVR().Version,
@@ -42,8 +44,8 @@ func Install(name string, opts ...CfgFn) feature.StepFn {
 		fn(cfg)
 	}
 	return func(ctx context.Context, t feature.T) {
-		if _, err := manifest.InstallLocalYaml(ctx, cfg); err != nil {
-			t.Fatal(err, cfg)
+		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+			t.Fatal(err)
 		}
 	}
 }
@@ -54,7 +56,7 @@ func IsReady(name string, timings ...time.Duration) feature.StepFn {
 }
 
 // WithVersion overrides the default API version
-func WithVersion(version string) CfgFn {
+func WithVersion(version string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if version != "" {
 			cfg["version"] = version
@@ -63,14 +65,14 @@ func WithVersion(version string) CfgFn {
 }
 
 // WithNumPartitions adds the numPartitions config to a KafkaChannel spec.
-func WithNumPartitions(numPartitions string) CfgFn {
+func WithNumPartitions(numPartitions string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		cfg["numPartitions"] = numPartitions
 	}
 }
 
 // WithReplicationFactor adds the replicationFactor config to a KafkaChannel spec.
-func WithReplicationFactor(replicationFactor string) CfgFn {
+func WithReplicationFactor(replicationFactor string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		cfg["replicationFactor"] = replicationFactor
 	}
diff --git a/test/rekt/resources/kafkachannel/kafkachannel_test.go b/test/rekt/resources/kafkachannel/kafkachannel_test.go
index 4bef7782c..7142cf87a 100644
--- a/test/rekt/resources/kafkachannel/kafkachannel_test.go
+++ b/test/rekt/resources/kafkachannel/kafkachannel_test.go
@@ -34,14 +34,7 @@ func Example() {
 	WithNumPartitions("10")(cfg)
 	WithReplicationFactor("3")(cfg)
 
-	directory, err := os.Getwd()
-	if err != nil {
-		panic(err)
-	}
-
-	dirFS := os.DirFS(directory)
-
-	files, err := manifest.ExecuteYAML(dirFS, images, cfg)
+	files, err := manifest.ExecuteYAML(yaml, images, cfg)
 	if err != nil {
 		panic(err)
 	}
diff --git a/test/rekt/resources/kafkasource/kafkasource.go b/test/rekt/resources/kafkasource/kafkasource.go
index 247f57ffb..5e903af3d 100644
--- a/test/rekt/resources/kafkasource/kafkasource.go
+++ b/test/rekt/resources/kafkasource/kafkasource.go
@@ -18,6 +18,7 @@ package kafkasource
 
 import (
 	"context"
+	"embed"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -27,14 +28,15 @@ import (
 	"knative.dev/reconciler-test/pkg/manifest"
 )
 
-type CfgFn func(map[string]interface{})
+//go:embed *.yaml
+var yaml embed.FS
 
 func GVR() schema.GroupVersionResource {
 	return schema.GroupVersionResource{Group: "sources.knative.dev", Version: "v1beta1", Resource: "kafkasources"}
 }
 
 // Install will create a KafkaSource resource, using the latest version, augmented with the config fn options.
-func Install(name string, opts ...CfgFn) feature.StepFn {
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 	cfg := map[string]interface{}{
 		"name":    name,
 		"version": GVR().Version,
@@ -55,7 +57,7 @@ func IsReady(name string, timings ...time.Duration) feature.StepFn {
 }
 
 // WithVersion overrides the default API version
-func WithVersion(version string) CfgFn {
+func WithVersion(version string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if version != "" {
 			cfg["version"] = version
@@ -64,7 +66,7 @@ func WithVersion(version string) CfgFn {
 }
 
 // WithAnnotations adds annotation to a KafkaSource metadata.
-func WithAnnotations(annotations map[string]string) CfgFn {
+func WithAnnotations(annotations map[string]string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if annotations != nil {
 			cfg["annotations"] = annotations
@@ -73,7 +75,7 @@ func WithAnnotations(annotations map[string]string) CfgFn {
 }
 
 // WithBootstrapServers adds the bootstrapServers config to a KafkaSource spec.
-func WithBootstrapServers(bootstrapServers []string) CfgFn {
+func WithBootstrapServers(bootstrapServers []string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if bootstrapServers != nil {
 			cfg["bootstrapServers"] = bootstrapServers
@@ -82,7 +84,7 @@ func WithBootstrapServers(bootstrapServers []string) CfgFn {
 }
 
 // WithTopics adds the topics config to a KafkaSource spec.
-func WithTopics(topics []string) CfgFn {
+func WithTopics(topics []string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if topics != nil {
 			cfg["topics"] = topics
@@ -91,7 +93,7 @@ func WithTopics(topics []string) CfgFn {
 }
 
 // WithTLSEnabled enables TLS to a KafkaSource spec.
-func WithTLSEnabled() CfgFn {
+func WithTLSEnabled() manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, ok := cfg["tls"]; !ok {
 			cfg["tls"] = map[string]interface{}{}
@@ -100,7 +102,7 @@ func WithTLSEnabled() CfgFn {
 }
 
 // WithTLSCert adds the TLS cert config to a KafkaSource spec.
-func WithTLSCert(name, key string) CfgFn {
+func WithTLSCert(name, key string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, ok := cfg["tls"]; !ok {
 			cfg["tls"] = map[string]interface{}{}
@@ -116,7 +118,7 @@ func WithTLSCert(name, key string) CfgFn {
 }
 
 // WithTLSKey adds the TLS key config to a KafkaSource spec.
-func WithTLSKey(name, key string) CfgFn {
+func WithTLSKey(name, key string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, ok := cfg["tls"]; !ok {
 			cfg["tls"] = map[string]interface{}{}
@@ -132,7 +134,7 @@ func WithTLSKey(name, key string) CfgFn {
 }
 
 // WithTLSCACert adds the TLS caCert config to a KafkaSource spec.
-func WithTLSCACert(name, key string) CfgFn {
+func WithTLSCACert(name, key string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, ok := cfg["tls"]; !ok {
 			cfg["tls"] = map[string]interface{}{}
@@ -148,7 +150,7 @@ func WithTLSCACert(name, key string) CfgFn {
 }
 
 // WithSASLEnabled enables SASL to a KafkaSource spec.
-func WithSASLEnabled() CfgFn {
+func WithSASLEnabled() manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, ok := cfg["sasl"]; !ok {
 			cfg["sasl"] = map[string]interface{}{}
@@ -157,7 +159,7 @@ func WithSASLEnabled() CfgFn {
 }
 
 // WithSASLUser adds the SASL user config to a KafkaSource spec.
-func WithSASLUser(name, key string) CfgFn {
+func WithSASLUser(name, key string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, ok := cfg["sasl"]; !ok {
 			cfg["sasl"] = map[string]interface{}{}
@@ -173,7 +175,7 @@ func WithSASLUser(name, key string) CfgFn {
 }
 
 // WithSASLPassword adds the SASL password config to a KafkaSource spec.
-func WithSASLPassword(name, key string) CfgFn {
+func WithSASLPassword(name, key string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, ok := cfg["sasl"]; !ok {
 			cfg["sasl"] = map[string]interface{}{}
@@ -189,7 +191,7 @@ func WithSASLPassword(name, key string) CfgFn {
 }
 
 // WithSASLType adds the SASL type config to a KafkaSource spec.
-func WithSASLType(name, key string) CfgFn {
+func WithSASLType(name, key string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, ok := cfg["sasl"]; !ok {
 			cfg["sasl"] = map[string]interface{}{}
@@ -205,7 +207,7 @@ func WithSASLType(name, key string) CfgFn {
 }
 
 // WithSink adds the sink related config to a PingSource spec.
-func WithSink(ref *duckv1.KReference, uri string) CfgFn {
+func WithSink(ref *duckv1.KReference, uri string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, set := cfg["sink"]; !set {
 			cfg["sink"] = map[string]interface{}{}
diff --git a/test/rekt/resources/kafkasource/kafkasource_test.go b/test/rekt/resources/kafkasource/kafkasource_test.go
index 48e4ae24e..f021f45bb 100644
--- a/test/rekt/resources/kafkasource/kafkasource_test.go
+++ b/test/rekt/resources/kafkasource/kafkasource_test.go
@@ -34,7 +34,7 @@ func Example_min() {
 		"version":   "v1beta1",
 	}
 
-	files, err := manifest.ExecuteLocalYAML(images, cfg)
+	files, err := manifest.ExecuteYAML(yaml, images, cfg)
 	if err != nil {
 		panic(err)
 	}
diff --git a/test/rekt/resources/kafkatopic/topic.go b/test/rekt/resources/kafkatopic/topic.go
index bebc4c9ab..858065a19 100644
--- a/test/rekt/resources/kafkatopic/topic.go
+++ b/test/rekt/resources/kafkatopic/topic.go
@@ -18,6 +18,7 @@ package kafkatopic
 
 import (
 	"context"
+	"embed"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -32,14 +33,15 @@ const (
 	kafkaNamespace = "kafka"
 )
 
-type CfgFn func(map[string]interface{})
+//go:embed *.yaml
+var yaml embed.FS
 
 func GVR() schema.GroupVersionResource {
 	return schema.GroupVersionResource{Group: "kafka.strimzi.io", Version: "v1beta2", Resource: "kafkatopics"}
 }
 
 // Install will create a Kafka Topic via the Strimzi topic CRD,, augmented with the config fn options.
-func Install(name string, opts ...CfgFn) feature.StepFn {
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 	cfg := map[string]interface{}{
 		"name":             name,
 		"clusterNamespace": "kafka",
@@ -68,21 +70,21 @@ func IsReady(name string, timings ...time.Duration) feature.StepFn {
 }
 
 // WithPartitions overrides the number of partitions (default: 10).
-func WithPartitions(partitions string) CfgFn {
+func WithPartitions(partitions string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		cfg["partitions"] = partitions
 	}
 }
 
 // WithClusterName overrides the Kafka cluster names where to create the topic (default: my-cluster)
-func WithClusterName(name string) CfgFn {
+func WithClusterName(name string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		cfg["clusterName"] = name
 	}
 }
 
 // WithClusterNamespace overrides the Kafka cluster namespace where to create the topic (default: kafka)
-func WithClusterNamespace(namespace string) CfgFn {
+func WithClusterNamespace(namespace string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		cfg["clusterNamespace"] = namespace
 	}
diff --git a/test/rekt/resources/kafkatopic/topic_test.go b/test/rekt/resources/kafkatopic/topic_test.go
index e7c60c8bf..baa95f24d 100644
--- a/test/rekt/resources/kafkatopic/topic_test.go
+++ b/test/rekt/resources/kafkatopic/topic_test.go
@@ -31,7 +31,7 @@ func Example_min() {
 		"clusterNamespace": "there",
 	}
 
-	files, err := manifest.ExecuteLocalYAML(images, cfg)
+	files, err := manifest.ExecuteYAML(yaml, images, cfg)
 	if err != nil {
 		panic(err)
 	}
diff --git a/test/rekt/resources/resetoffset/resetoffset.go b/test/rekt/resources/resetoffset/resetoffset.go
index 186075dcd..571b8b639 100644
--- a/test/rekt/resources/resetoffset/resetoffset.go
+++ b/test/rekt/resources/resetoffset/resetoffset.go
@@ -18,6 +18,7 @@ package resetoffset
 
 import (
 	"context"
+	"embed"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -28,14 +29,15 @@ import (
 	"knative.dev/reconciler-test/pkg/manifest"
 )
 
-type CfgFn func(map[string]interface{})
+//go:embed *.yaml
+var yaml embed.FS
 
 func GVR() schema.GroupVersionResource {
 	return schema.GroupVersionResource{Group: "kafka.eventing.knative.dev", Version: "v1alpha1", Resource: "resetoffsets"}
 }
 
 // Install will create a ResetOffset resource, using the latest version, augmented with the config fn options.
-func Install(name string, opts ...CfgFn) feature.StepFn {
+func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 	cfg := map[string]interface{}{
 		"name":    name,
 		"version": GVR().Version,
@@ -62,7 +64,7 @@ func IsSucceeded(name string, timings ...time.Duration) feature.StepFn {
 }
 
 // WithVersion overrides the default API version
-func WithVersion(version string) CfgFn {
+func WithVersion(version string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if version != "" {
 			cfg["version"] = version
@@ -71,7 +73,7 @@ func WithVersion(version string) CfgFn {
 }
 
 // WithOffsetTime adds the offsetTime config to a ResetOffset spec.
-func WithOffsetTime(offsetTime string) CfgFn {
+func WithOffsetTime(offsetTime string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		offsetCfg := map[string]interface{}{}
 		offsetCfg["time"] = offsetTime
@@ -80,7 +82,7 @@ func WithOffsetTime(offsetTime string) CfgFn {
 }
 
 // WithRef adds the ref config to a ResetOffset spec.
-func WithRef(ref *duckv1.KReference) CfgFn {
+func WithRef(ref *duckv1.KReference) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		refCfg := map[string]interface{}{}
 		refCfg["name"] = ref.Name
diff --git a/test/rekt/resources/resetoffset/resetoffset_test.go b/test/rekt/resources/resetoffset/resetoffset_test.go
index 6e5964287..32977041d 100644
--- a/test/rekt/resources/resetoffset/resetoffset_test.go
+++ b/test/rekt/resources/resetoffset/resetoffset_test.go
@@ -40,14 +40,7 @@ func Example() {
 		Name:       "baz",
 	})(cfg)
 
-	directory, err := os.Getwd()
-	if err != nil {
-		panic(err)
-	}
-
-	dirFS := os.DirFS(directory)
-
-	files, err := manifest.ExecuteYAML(dirFS, images, cfg)
+	files, err := manifest.ExecuteYAML(yaml, images, cfg)
 	if err != nil {
 		panic(err)
 	}
diff --git a/vendor/knative.dev/eventing/test/upgrade/prober/prober.go b/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
index 2c204eb91..bb8908684 100644
--- a/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
+++ b/vendor/knative.dev/eventing/test/upgrade/prober/prober.go
@@ -66,8 +66,6 @@ func (p *probeRunner) Verify(ctx pkgupgrade.Context) {
 	// use T from new test
 	p.client.T = ctx.T
 	t := ctx.T
-	defer testlib.TearDown(p.client)
-	defer p.remove()
 	p.Finish()
 	waitAfterFinished(p.prober)
 
@@ -80,6 +78,10 @@ func (p *probeRunner) Verify(ctx pkgupgrade.Context) {
 	}
 
 	p.ReportErrors(errors)
+	if !ctx.T.Failed() {
+		p.remove()
+		testlib.TearDown(p.client)
+	}
 }
 
 type prober struct {