Requeue at a fixed interval while the consumer group is not empty (#4828)

* Requeue at a fixed interval while the consumer group is not empty

Since #4785 the ConsumerGroup finalizer retries deleting a Kafka group
that still has members without limit. It returns an error, so the
controller requeues with the rate limiter, which doubles the delay from
5ms up to 1000s. A dispatcher stays in the group until its in-flight
delivery ends, which takes up to the 60s connect timeout when the
subscriber is gone. By then the finalizer has failed about 25 times and
its next retry is about 17 minutes away, so the owner (KafkaChannel,
Trigger, KafkaSource) stays until something else enqueues the key.

Return controller.NewRequeueAfter(5s) instead, as the channel, broker
and sink reconcilers do, so the finalizer retries within 5s of the group
becoming empty. The status still records the failure.

Seen in channel-reconciler-tests-sasl-ssl on #4825:
- 11:04:06-11:04:44 "The group is not empty", backoff 63ms..2.5s..
- 11:05:10 the dispatcher closed its consumer after a 60s connect timeout
  to the deleted subscriber
- 11:06:04 the test cleanup gave up with the Channel still present
- 11:07:44 the finalizer was removed, when a new leader took the lease

* Leave a consumer group in Kafka when it keeps members for 5 minutes

The ConsumerGroup finalizer retries while the Kafka group still has
members. A dispatcher leaves within about a minute, but a member that
Knative does not control never does, for example another application
using the same group.id as a KafkaSource. The finalizer then never
finishes, and neither does the deletion of the KafkaSource, Trigger or
KafkaChannel, or of their namespace.

Count the attempts with an expiring counter, as the broker finalizer does
for an unreachable Kafka cluster. After 60 attempts 5s apart, about 5
minutes, leave the group in Kafka, as the finalizer did before #4785, and
log it.

Author: gauron99

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

@@ -44,6 +44,7 @@ import (
 	"knative.dev/eventing/pkg/observability"
 	"knative.dev/eventing/pkg/scheduler"
 	"knative.dev/pkg/apis"
+	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/observability/attributekey"
 	pointer "knative.dev/pkg/ptr"
@@ -131,6 +132,10 @@ type Reconciler struct {
 	// tried to delete consumer group metadata from Kafka.
 	DeleteConsumerGroupMetadataCounter *counter.Counter
 
+	// NonEmptyGroupCounter is an in-memory counter to count how many times we have
+	// found the consumer group in Kafka still having members while deleting it.
+	NonEmptyGroupCounter *counter.Counter
+
 	// InitOffsetLatestInitialOffsetCache is the cache for consumer group offset initialization.
 	//
 	// When there is high load and multiple consumer group schedule calls, we get many
@@ -257,14 +262,26 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, cg *kafkainternals.Consum
 		// The group still has active members (dispatchers that haven't observed the contract
 		// update yet), so retry without consuming the bounded retry budget used for
 		// unreachable clusters.
+		// A member can stay until its in-flight event delivery ends, which takes up to the
+		// 60s connect timeout. Retry at a fixed interval: returning the error would back
+		// off exponentially, up to 1000s, and retry long after the group became empty.
 		if errorIsOneOf(err, sarama.ErrNonEmptyGroup) {
-			return cg.MarkDeleteOffsetFailed("DeleteConsumerGroupOffset", err)
-		}
-		// We retry a few times to delete Consumer group metadata from Kafka before giving up.
-		if v := r.DeleteConsumerGroupMetadataCounter.Inc(string(cg.GetUID())); v <= 5 {
-			return cg.MarkDeleteOffsetFailed("DeleteConsumerGroupOffset", fmt.Errorf("%w (retry num %d)", err, v))
+			// A member that stays for about 5 minutes is not a dispatcher on its way out,
+			// for example another application using the same group.id. Leave the group in
+			// Kafka rather than block the deletion of the owner forever.
+			if r.NonEmptyGroupCounter.Inc(string(cg.GetUID())) <= 60 {
+				_ = cg.MarkDeleteOffsetFailed("DeleteConsumerGroupOffset", err)
+				return controller.NewRequeueAfter(5 * time.Second)
+			}
+			r.NonEmptyGroupCounter.Del(string(cg.GetUID()))
+			logger.Errorw("Leaving the consumer group in Kafka, it still has members", zap.Error(err))
+		} else {
+			// We retry a few times to delete Consumer group metadata from Kafka before giving up.
+			if v := r.DeleteConsumerGroupMetadataCounter.Inc(string(cg.GetUID())); v <= 5 {
+				return cg.MarkDeleteOffsetFailed("DeleteConsumerGroupOffset", fmt.Errorf("%w (retry num %d)", err, v))
+			}
+			r.DeleteConsumerGroupMetadataCounter.Del(string(cg.GetUID()))
 		}
-		r.DeleteConsumerGroupMetadataCounter.Del(string(cg.GetUID()))
 	}
 
 	r.InitOffsetLatestInitialOffsetCache.Expire(keyOf(cg))

control-plane/pkg/reconciler/consumergroup/consumergroup_test.go

@@ -64,6 +64,7 @@ const (
 	testSchedulerKey           = "scheduler"
 	noTestScheduler            = "no-scheduler"
 	wantErrorOnGetClusterAdmin = "wantErrorOnGetClusterAdmin"
+	nonEmptyGroupAttemptsKey   = "nonEmptyGroupAttempts"
 
 	systemNamespace = "knative-eventing"
 	finalizerName   = "consumergroups.internal.kafka.eventing.knative.dev"
@@ -1881,6 +1882,7 @@ func TestReconcileKind(t *testing.T) {
 			SystemNamespace:                    systemNamespace,
 			AutoscalerConfig:                   "",
 			DeleteConsumerGroupMetadataCounter: counter.NewExpiringCounter(ctx),
+			NonEmptyGroupCounter:               counter.NewExpiringCounter(ctx),
 			InitOffsetLatestInitialOffsetCache: prober.NewLocalExpiringCache[string, prober.Status, struct{}](ctx, time.Second),
 			EnqueueKey:                         func(key string) {},
 		}
@@ -2041,6 +2043,7 @@ func TestReconcileKindNoAutoscaler(t *testing.T) {
 			},
 			SystemNamespace:                    systemNamespace,
 			DeleteConsumerGroupMetadataCounter: counter.NewExpiringCounter(ctx),
+			NonEmptyGroupCounter:               counter.NewExpiringCounter(ctx),
 			InitOffsetLatestInitialOffsetCache: prober.NewLocalExpiringCache[string, prober.Status, struct{}](ctx, time.Second),
 			EnqueueKey:                         func(key string) {},
 		}
@@ -2578,7 +2581,7 @@ func TestFinalizeKind(t *testing.T) {
 			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
 		},
 		{
-			Name: "Finalize error - failed consumer group deletion with " + sarama.ErrNonEmptyGroup.Error(),
+			Name: "Finalize requeue - failed consumer group deletion with " + sarama.ErrNonEmptyGroup.Error(),
 			Objects: []runtime.Object{
 				NewDeletedConsumeGroup(
 					ConsumerGroupOwnerRef(SourceAsOwnerReference()),
@@ -2598,13 +2601,6 @@ func TestFinalizeKind(t *testing.T) {
 				}),
 				kafkatesting.ErrorOnDeleteConsumerGroupTestKey: sarama.ErrNonEmptyGroup,
 			},
-			WantEvents: []string{
-				Eventf(
-					corev1.EventTypeWarning,
-					"InternalError",
-					"failed to delete consumer group offset: unable to delete the consumer group my.group.id: "+sarama.ErrNonEmptyGroup.Error(),
-				),
-			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: func() runtime.Object {
@@ -2626,6 +2622,29 @@ func TestFinalizeKind(t *testing.T) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
 		},
+		{
+			Name: "Finalize normal - consumer group still has members after 60 attempts",
+			Objects: []runtime.Object{
+				NewDeletedConsumeGroup(
+					ConsumerGroupOwnerRef(SourceAsOwnerReference()),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(ChannelBootstrapServers),
+							ConsumerGroupIDConfig("my.group.id"),
+						),
+					)),
+				),
+			},
+			Key: testKey,
+			OtherTestData: map[string]interface{}{
+				testSchedulerKey: SchedulerFunc(func(_ context.Context, vpod scheduler.VPod) ([]eventingduckv1alpha1.Placement, error) {
+					return nil, nil
+				}),
+				kafkatesting.ErrorOnDeleteConsumerGroupTestKey: sarama.ErrNonEmptyGroup,
+				nonEmptyGroupAttemptsKey:                       60,
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the source namespace with configmap namespace, so skip it
+		},
 	}
 
 	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
@@ -2662,6 +2681,7 @@ func TestFinalizeKind(t *testing.T) {
 			},
 			KafkaFeatureFlags:                  configapis.DefaultFeaturesConfig(),
 			DeleteConsumerGroupMetadataCounter: counter.NewExpiringCounter(ctx),
+			NonEmptyGroupCounter:               counter.NewExpiringCounter(ctx),
 			InitOffsetLatestInitialOffsetCache: prober.NewLocalExpiringCache[string, prober.Status, struct{}](ctx, time.Second),
 		}
 
@@ -2670,6 +2690,12 @@ func TestFinalizeKind(t *testing.T) {
 		r.ExpectedReplicas, _ = meter.Int64Gauge("kn.eventing.replicas.expected")
 		r.ReadyReplicas, _ = meter.Int64Gauge("kn.eventing.replicas.ready")
 
+		// Earlier attempts on the ConsumerGroup, which has no UID in these tests.
+		attempts, _ := row.OtherTestData[nonEmptyGroupAttemptsKey].(int)
+		for i := 0; i < attempts; i++ {
+			r.NonEmptyGroupCounter.Inc("")
+		}
+
 		return consumergroup.NewReconciler(
 			ctx,
 			logging.FromContext(ctx),

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

@@ -139,6 +139,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher) *controller.I
 		KedaClient:                         kedaclient.Get(ctx),
 		AutoscalerConfig:                   env.AutoscalerConfigMap,
 		DeleteConsumerGroupMetadataCounter: counter.NewExpiringCounter(ctx),
+		NonEmptyGroupCounter:               counter.NewExpiringCounter(ctx),
 		InitOffsetLatestInitialOffsetCache: prober.NewLocalExpiringCache[string, prober.Status, struct{}](ctx, 20*time.Minute),
 	}