Repository: knative-sandbox/eventing-kafka-broker
Pull request #2183 (2022-05-11 09:29:28)

[release-1.2] Handle committing offsets on revoked partitions
This is an automated cherry-pick of #2179

knative-prow[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2183#" title="Author self-approved">knative-prow-robot</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **julz** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @julz` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.2/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["julz"]} -->

Patch:
diff --git a/control-plane/cmd/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
index 5447ae63e..5a9624445 100644
--- a/control-plane/cmd/kafka-controller/main.go
+++ b/control-plane/cmd/kafka-controller/main.go
@@ -83,7 +83,7 @@ func main() {
 		injection.NamedControllerConstructor{
 			Name: "channel-controller",
 			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-				return channel.NewController(ctx, watcher, channelEnv)
+				return channel.NewController(ctx, channelEnv)
 			},
 		},
 
diff --git a/control-plane/pkg/prober/async_prober.go b/control-plane/pkg/prober/async_prober.go
index 1d6d2ed6d..c4f211c09 100644
--- a/control-plane/pkg/prober/async_prober.go
+++ b/control-plane/pkg/prober/async_prober.go
@@ -22,57 +22,55 @@ import (
 	"time"
 
 	"go.uber.org/zap"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
-	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	"knative.dev/pkg/logging"
 )
 
+type IPsLister func() ([]string, error)
+
 type asyncProber struct {
 	client    httpClient
 	enqueue   EnqueueFunc
 	logger    *zap.Logger
 	cache     Cache
-	podLister func() ([]*corev1.Pod, error)
+	IPsLister IPsLister
 	port      string
 }
 
 // NewAsync creates an async Prober.
 //
 // It reports status changes using the provided EnqueueFunc.
-func NewAsync(ctx context.Context, client httpClient, port string, podsLabelsSelector labels.Selector, enqueue EnqueueFunc) Prober {
+func NewAsync(ctx context.Context, client httpClient, port string, IPsLister IPsLister, enqueue EnqueueFunc) Prober {
 	logger := logging.FromContext(ctx).Desugar().
 		With(zap.String("scope", "prober"))
 
 	if len(port) == 0 {
 		logger.Fatal("Port is required")
 	}
-	podLister := podinformer.Get(ctx).Lister()
 	return &asyncProber{
 		client:    client,
 		enqueue:   enqueue,
 		logger:    logger,
 		cache:     NewLocalExpiringCache(ctx, 30*time.Minute),
-		podLister: func() ([]*corev1.Pod, error) { return podLister.List(podsLabelsSelector) },
+		IPsLister: IPsLister,
 		port:      port,
 	}
 }
 
 func (a *asyncProber) Probe(ctx context.Context, addressable Addressable, expected Status) Status {
 	address := addressable.Address
-	pods, err := a.podLister()
+	IPs, err := a.IPsLister()
 	if err != nil {
-		a.logger.Error("Failed to list pods", zap.Error(err))
+		a.logger.Error("Failed to list IPs", zap.Error(err))
 		return StatusUnknown
 	}
-	// Return `StatusNotReady` when there are no pods.
-	if len(pods) == 0 {
+	// Return `StatusNotReady` when there are no IPs.
+	if len(IPs) == 0 {
 		return StatusNotReady
 	}
 
 	// aggregatedCurrentKnownStatus keeps track of the current status in the cache excluding `StatusReady`
-	// since we just skip pods that have `StatusReady`.
+	// since we just skip IPs that have `StatusReady`.
 	//
 	// If there is a status that is `StatusUnknown` the final status  we want to return is `StatusUnknown`,
 	// while we return `StatusNotReady` when the status is known and all probes returned `StatusNotReady`.
@@ -83,18 +81,18 @@ func (a *asyncProber) Probe(ctx context.Context, addressable Addressable, expect
 	// It goes to done once we have all probe request results regardless of whether they are coming from
 	// the cache or from an actual request.
 	var wg sync.WaitGroup
-	wg.Add(len(pods))
+	wg.Add(len(IPs))
 
 	// enqueueOnce allows requeuing the resource only once, when we have all probe request results.
 	var enqueueOnce sync.Once
 
-	for _, p := range pods {
+	for _, IP := range IPs {
 		podUrl := *address
-		podUrl.Host = p.Status.PodIP + ":" + a.port
+		podUrl.Host = IP + ":" + a.port
 		address := podUrl.String()
 
 		logger := a.logger.
-			With(zap.String("pod.metadata.name", p.Name)).
+			With(zap.String("IP", IP)).
 			With(zap.String("address", address))
 
 		currentStatus := a.cache.GetStatus(address)
diff --git a/control-plane/pkg/prober/prober.go b/control-plane/pkg/prober/prober.go
index 857693d9c..b0cf2f7d8 100644
--- a/control-plane/pkg/prober/prober.go
+++ b/control-plane/pkg/prober/prober.go
@@ -18,6 +18,7 @@ package prober
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"net/url"
 
@@ -89,3 +90,9 @@ func probe(ctx context.Context, client httpClient, logger *zap.Logger, address s
 
 	return StatusReady
 }
+
+func IPsListerFromService(svc types.NamespacedName) IPsLister {
+	return func() ([]string, error) {
+		return []string{fmt.Sprintf("%s.%s.svc", svc.Name, svc.Namespace)}, nil
+	}
+}
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index d488bee56..2c3dbbb9a 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -24,6 +24,7 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
@@ -88,7 +89,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 	})
 
 	reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
-	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, env.IngressPodPort, reconciler.ReceiverSelector(), impl.EnqueueKey)
+	IPsLister := prober.IPsListerFromService(types.NamespacedName{Namespace: env.SystemNamespace, Name: env.IngressName})
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, env.IngressPodPort, IPsLister, impl.EnqueueKey)
 	reconciler.IngressHost = network.GetServiceHostname(env.IngressName, env.SystemNamespace)
 
 	brokerInformer := brokerinformer.Get(ctx)
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index 8ccfa800b..febb5bfc1 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -23,6 +23,7 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
@@ -35,7 +36,6 @@ import (
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 
-	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/network"
@@ -46,8 +46,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
 
-// TODO: no need for a configmap watcher?
-func NewController(ctx context.Context, watcher configmap.Watcher, configs *config.Env) *controller.Impl {
+func NewController(ctx context.Context, configs *config.Env) *controller.Impl {
 
 	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
 
@@ -83,7 +82,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 	}
 
 	impl := kafkachannelreconciler.NewImpl(ctx, reconciler)
-	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, reconciler.ReceiverSelector(), impl.EnqueueKey)
+	IPsLister := prober.IPsListerFromService(types.NamespacedName{Namespace: configs.SystemNamespace, Name: configs.IngressName})
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, IPsLister, impl.EnqueueKey)
 	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
 
 	channelInformer := kafkachannelinformer.Get(ctx)
diff --git a/control-plane/pkg/reconciler/sink/controller.go b/control-plane/pkg/reconciler/sink/controller.go
index 967dcf4b6..b0de02839 100644
--- a/control-plane/pkg/reconciler/sink/controller.go
+++ b/control-plane/pkg/reconciler/sink/controller.go
@@ -23,6 +23,7 @@ import (
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
@@ -75,7 +76,8 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 	}
 
 	impl := sinkreconciler.NewImpl(ctx, reconciler)
-	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, reconciler.ReceiverSelector(), impl.EnqueueKey)
+	IPsLister := prober.IPsListerFromService(types.NamespacedName{Namespace: configs.SystemNamespace, Name: configs.IngressName})
+	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, configs.IngressPodPort, IPsLister, impl.EnqueueKey)
 	reconciler.IngressHost = network.GetServiceHostname(configs.IngressName, configs.SystemNamespace)
 
 	sinkInformer := sinkinformer.Get(ctx)
diff --git a/data-plane/config/broker/500-receiver.yaml b/data-plane/config/broker/500-receiver.yaml
index 7d0ff7262..cdc550d49 100644
--- a/data-plane/config/broker/500-receiver.yaml
+++ b/data-plane/config/broker/500-receiver.yaml
@@ -174,6 +174,10 @@ spec:
       port: 80
       protocol: TCP
       targetPort: 8080
+    - name: http-container
+      port: 8080
+      protocol: TCP
+      targetPort: 8080
     - name: http-metrics
       port: 9090
       protocol: TCP
diff --git a/data-plane/config/channel/500-receiver.yaml b/data-plane/config/channel/500-receiver.yaml
index 317f693a9..88b2cf472 100644
--- a/data-plane/config/channel/500-receiver.yaml
+++ b/data-plane/config/channel/500-receiver.yaml
@@ -174,6 +174,10 @@ spec:
       port: 80
       protocol: TCP
       targetPort: 8080
+    - name: http-container
+      port: 8080
+      protocol: TCP
+      targetPort: 8080
     - name: http-metrics
       port: 9090
       protocol: TCP
diff --git a/data-plane/config/sink/500-receiver.yaml b/data-plane/config/sink/500-receiver.yaml
index 248672c44..b2b34762b 100644
--- a/data-plane/config/sink/500-receiver.yaml
+++ b/data-plane/config/sink/500-receiver.yaml
@@ -174,6 +174,10 @@ spec:
       port: 80
       protocol: TCP
       targetPort: 8080
+    - name: http-container
+      port: 8080
+      protocol: TCP
+      targetPort: 8080
     - name: http-metrics
       port: 9090
       protocol: TCP
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
index e49cdec82..ab07e69f2 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OffsetManager.java
@@ -113,9 +113,14 @@ public final class OffsetManager implements RecordDispatcherListener {
   }
 
   private void commit(final KafkaConsumerRecord<?, ?> record) {
-    this.offsetTrackers
-      .get(new TopicPartition(record.topic(), record.partition()))
-      .recordNewOffset(record.offset());
+    // We need to handle the case when the offset tracker was removed from our Map since when partitions are revoked we
+    // remove the associated offset tracker, however, we may get a response from the sink for a previously owned
+    // partition after a partition has been revoked.
+    // Note: it's not possible to commit offsets of partitions that this a particular consumer instance doesn't own.
+    final var ot = this.offsetTrackers.get(new TopicPartition(record.topic(), record.partition()));
+    if (ot != null) {
+      ot.recordNewOffset(record.offset());
+    }
   }
 
   private synchronized Future<Void> commit(final TopicPartition topicPartition, final OffsetTracker tracker) {
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
index ba9caf8e8..280727369 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSender.java
@@ -142,6 +142,7 @@ public final class WebClientCloudEventSender implements CloudEventSender {
     // From https://github.com/knative/specs/blob/c348f501de9eb998b4fd010c54d9127033ee41be/specs/eventing/data-plane.md#event-acknowledgement-and-delivery-retry
     return statusCode >= 500 || // Generic error
       statusCode == 404 || // Endpoint does not exist
+      statusCode == 408 || // Request Timeout
       statusCode == 409 || // Conflict / Processing in progress
       statusCode == 429;   // Too Many Requests / Overloaded
   }


Test patch:
diff --git a/control-plane/pkg/prober/async_prober_test.go b/control-plane/pkg/prober/async_prober_test.go
index d68fe4481..893a3b6c8 100644
--- a/control-plane/pkg/prober/async_prober_test.go
+++ b/control-plane/pkg/prober/async_prober_test.go
@@ -132,7 +132,18 @@ func TestAsyncProber(t *testing.T) {
 			u, _ := url.Parse(s.URL)
 
 			wantRequeueCountMin := atomic.NewInt64(int64(tc.wantRequeueCountMin))
-			prober := NewAsync(ctx, s.Client(), u.Port(), tc.podsLabelsSelector, func(key types.NamespacedName) {
+			var IPsLister IPsLister = func() ([]string, error) {
+				pods, err := podinformer.Get(ctx).Lister().List(tc.podsLabelsSelector)
+				if err != nil {
+					return nil, err
+				}
+				ips := make([]string, 0, len(pods))
+				for _, p := range pods {
+					ips = append(ips, p.Status.PodIP)
+				}
+				return ips, nil
+			}
+			prober := NewAsync(ctx, s.Client(), u.Port(), IPsLister, func(key types.NamespacedName) {
 				wantRequeueCountMin.Dec()
 			})
 
diff --git a/control-plane/pkg/prober/prober_test.go b/control-plane/pkg/prober/prober_test.go
index bb9bdf30b..596c8985d 100644
--- a/control-plane/pkg/prober/prober_test.go
+++ b/control-plane/pkg/prober/prober_test.go
@@ -20,8 +20,10 @@ import (
 	"context"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 func TestFuncProbe(t *testing.T) {
@@ -38,3 +40,33 @@ func TestFuncProbe(t *testing.T) {
 	require.Equal(t, status, s, s.String())
 	require.Equal(t, int32(1), calls.Load())
 }
+
+func TestIPsListerFromService(t *testing.T) {
+	tests := []struct {
+		name    string
+		svc     types.NamespacedName
+		want    []string
+		wantErr bool
+	}{
+		{
+			name: "ok",
+			svc: types.NamespacedName{
+				Namespace: "ns",
+				Name:      "name",
+			},
+			want:    []string{"name.ns.svc"},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := IPsListerFromService(tt.svc)()
+			if tt.wantErr != (err != nil) {
+				t.Errorf("Got err %v, wantErr %v", err, tt.wantErr)
+			}
+			if diff := cmp.Diff(tt.want, got); diff != "" {
+				t.Error("(-want, +got)", diff)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index a888eee34..ec3c090cf 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -72,23 +72,15 @@ var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
 	DataPlaneConfigMapName:      "kafka-channel-channels-subscriptions",
 	GeneralConfigMapName:        "kafka-channel-config",
-	IngressName:                 "kafka-channel-ingress",
+	IngressName:                 "kafka-channel-dispatcher",
 	SystemNamespace:             "knative-eventing",
 	DataPlaneConfigFormat:       base.Json,
 }
 
-// TODO: compare with inmemorychannel_test.go at /Users/aliok/code/github.com/knative/eventing/pkg/reconciler/inmemorychannel/controller/inmemorychannel_test.go
-// TODO: compare with channel_test.go at /Users/aliok/code/github.com/knative/eventing/pkg/reconciler/channel/channel_test.go
-// TODO: are we setting a InitialOffsetsCommitted status? we gotta set it on the subscription. 1) can we do it? 2) does it make sense?
-// TODO: test if things from channel spec is used properly?
-// TODO: test if things from subscription spec is used properly?
-// TODO: rename test cases consistently
-// TODO: test finalize?
-// TODO: fix the order of status updates
-// TODO: add tests where there are no CM updates
-
 func TestReconcileKind(t *testing.T) {
 
+	t.Setenv("SYSTEM_NAMESPACE", "knative-eventing")
+
 	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
 
 	env := *DefaultEnv
@@ -198,7 +190,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						ChannelAddressable(&env),
 						StatusProbeSucceeded,
 					),
@@ -267,7 +258,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						StatusProbeFailed(prober.StatusNotReady),
 					),
 				},
@@ -338,7 +328,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						StatusProbeFailed(prober.StatusUnknown),
 					),
 				},
@@ -409,7 +398,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						WithSubscribers(Subscriber1(WithFreshSubscriber, WithNoSubscriberURI)),
 					),
 				},
@@ -487,7 +475,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						ChannelAddressable(&env),
 						WithSubscribers(Subscriber1()),
 						StatusProbeSucceeded,
@@ -562,7 +549,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						ChannelAddressable(&env),
 						WithSubscribers(Subscriber1()),
 						StatusProbeSucceeded,
@@ -637,7 +623,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						ChannelAddressable(&env),
 						WithSubscribers(Subscriber1()),
 						StatusProbeSucceeded,
@@ -721,7 +706,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						ChannelAddressable(&env),
 						WithSubscribers(Subscriber1(), Subscriber2()),
 						StatusProbeSucceeded,
@@ -994,7 +978,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						ChannelAddressable(&env),
 						StatusProbeSucceeded,
 					),
@@ -1078,7 +1061,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						ChannelAddressable(&env),
 						WithSubscribers(Subscriber1()),
 						StatusProbeSucceeded,
@@ -1148,7 +1130,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						ChannelAddressable(&env),
 						StatusProbeSucceeded,
 					),
@@ -1215,7 +1196,6 @@ func TestReconcileKind(t *testing.T) {
 						StatusConfigMapUpdatedReady(&env),
 						StatusTopicReadyWithName(ChannelTopic()),
 						StatusDataPlaneAvailable,
-						//StatusInitialOffsetsCommitted,
 						ChannelAddressable(&env),
 						StatusProbeSucceeded,
 					),
diff --git a/control-plane/pkg/reconciler/channel/controller_test.go b/control-plane/pkg/reconciler/channel/controller_test.go
index cc3ca16ce..f7a2e0d08 100644
--- a/control-plane/pkg/reconciler/channel/controller_test.go
+++ b/control-plane/pkg/reconciler/channel/controller_test.go
@@ -29,7 +29,6 @@ import (
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
-	"knative.dev/pkg/configmap"
 	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
@@ -61,11 +60,6 @@ func TestNewController(t *testing.T) {
 
 	controller := NewController(
 		ctx,
-		configmap.NewStaticWatcher(&corev1.ConfigMap{
-			ObjectMeta: metav1.ObjectMeta{
-				Name: "cm",
-			},
-		}),
 		configs,
 	)
 	if controller == nil {
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
index 72ee6c55b..5042fdac2 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/http/WebClientCloudEventSenderTest.java
@@ -76,7 +76,7 @@ public class WebClientCloudEventSenderTest {
     return Stream.concat(
       Stream.concat(
         IntStream.range(200, 404).boxed(),
-        IntStream.range(405, 409).boxed()
+        IntStream.range(405, 408).boxed()
       ),
       Stream.concat(
         IntStream.range(410, 429).boxed(),
@@ -88,7 +88,7 @@ public class WebClientCloudEventSenderTest {
   public static Stream<Integer> retryableStatusCodes() {
     return Stream.concat(
       IntStream.range(500, 600).boxed(),
-      IntStream.of(404, 409, 429).boxed()
+      IntStream.of(404, 408, 409, 429).boxed()
     );
   }
 }