Repository: knative-sandbox/eventing-kafka-broker
Pull request #2160 (2022-05-05 15:19:32)

adding 408 - Request Timeout to list of retryable status codes
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

Fixes https://github.com/knative/specs/issues/96

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- adding `408` to the list of retries

<!--
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

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2160#" title="Author self-approved">matzew</a>*

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).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)~~ [matzew]

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":[]} -->
knative-prow[bot]: @matzew: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_main | d2f52230bf0dc3c62a91c2c1ca6d504521d6afbc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2160/reconciler-tests_eventing-kafka-broker_main/1522234552578215936) | true | `/test reconciler-tests_eventing-kafka-broker_main`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/control-plane/pkg/security/secret.go b/control-plane/pkg/security/secret.go
index 4dd8c7d82..dcdf67a1e 100644
--- a/control-plane/pkg/security/secret.go
+++ b/control-plane/pkg/security/secret.go
@@ -96,12 +96,12 @@ func secretData(data map[string][]byte) kafka.ConfigOption {
 func saslConfig(protocol string, data map[string][]byte) kafka.ConfigOption {
 	return func(config *sarama.Config) error {
 
-		// Supported mechanism SASL/PLAIN or SASL/SCRAM.
+		// Supported mechanism SASL/PLAIN (default if not specified) or SASL/SCRAM.
+		saslMechanism := SaslPlain
 		givenSASLMechanism, ok := data[SaslMechanismKey]
-		if !ok {
-			return fmt.Errorf("[protocol %s] SASL mechanism required (key: %s)", protocol, SaslMechanismKey)
+		if ok {
+			saslMechanism = string(givenSASLMechanism)
 		}
-		saslMechanism := string(givenSASLMechanism)
 
 		user, ok := data[SaslUserKey]
 		if !ok || len(user) == 0 {
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java
index 5db7ad77a..5d5d989bd 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutor.java
@@ -16,6 +16,7 @@
 package dev.knative.eventing.kafka.broker.core;
 
 import io.vertx.core.Future;
+
 import java.util.ArrayDeque;
 import java.util.Queue;
 import java.util.function.Supplier;
@@ -30,10 +31,12 @@ public class OrderedAsyncExecutor {
   private final Queue<Supplier<Future<?>>> queue;
 
   private boolean isStopped;
+  private boolean inFlight;
 
   public OrderedAsyncExecutor() {
     this.queue = new ArrayDeque<>();
     this.isStopped = false;
+    this.inFlight = false;
   }
 
   /**
@@ -42,38 +45,42 @@ public class OrderedAsyncExecutor {
    * @param task the task to offer
    */
   public void offer(Supplier<Future<?>> task) {
-    boolean wasEmpty = queue.isEmpty();
-    queue.offer(task);
+    if (this.isStopped) {
+      // Executor is stopped, return without adding the task to the queue.
+      return;
+    }
+    boolean wasEmpty = this.queue.isEmpty();
+    this.queue.offer(task);
     if (wasEmpty) { // If no elements in the queue, then we need to start consuming it
       consume();
     }
   }
 
-  /**
-   * Stop the executor. This won't stop the actual task on-fly, but it will prevent queued tasks to be executed.
-   */
-  public void stop() {
-    this.isStopped = true;
-  }
-
-  void consume() {
-    if (this.isStopped) {
+  private void consume() {
+    if (queue.isEmpty() || this.inFlight || this.isStopped) {
       return;
     }
-    Supplier<Future<?>> task = this.queue.peek();
-    if (task == null) {
-      return; // No task to process
-    }
-    task.get()
+    this.inFlight = true;
+    this.queue
+      .remove()
+      .get()
       .onComplete(ar -> {
-        // We don't actually care about the result,
-        // the task should have the failure handling by itself
-
-        // Remove the element from the queue
-        queue.poll();
-
+        this.inFlight = false;
         consume();
       });
   }
 
+  public boolean isWaitingForTasks() {
+    // TODO To perform this flag would be nice to take into account also the time that it takes for a sink to process a
+    //  message so that we can fetch records in advance and keep queues busy.
+    return this.queue.isEmpty();
+  }
+
+  /**
+   * Stop the executor. This won't stop the actual in-flight task, but it will prevent queued tasks to be executed.
+   */
+  public void stop() {
+    this.isStopped = true;
+    this.queue.clear();
+  }
 }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
index 9052ea892..5a63f663f 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
@@ -161,9 +161,13 @@ public class Metrics {
    * @return A meter binder to close once the consumer is closed.
    */
   public static <K, V> AutoCloseable register(final Consumer<K, V> consumer) {
-    final var clientMetrics = new KafkaClientMetrics(consumer);
-    clientMetrics.bindTo(getRegistry());
-    return clientMetrics;
+    final var registry = getRegistry();
+    if (registry != null) {
+      final var clientMetrics = new KafkaClientMetrics(consumer);
+      clientMetrics.bindTo(registry);
+      return clientMetrics;
+    }
+    return () -> {};
   }
 
   /**
@@ -175,9 +179,13 @@ public class Metrics {
    * @return A meter binder to close once the producer is closed.
    */
   public static <K, V> AutoCloseable register(final Producer<K, V> producer) {
-    final var clientMetrics = new KafkaClientMetrics(producer);
-    clientMetrics.bindTo(getRegistry());
-    return clientMetrics;
+    final var registry = getRegistry();
+    if (registry != null) {
+      final var clientMetrics = new KafkaClientMetrics(producer);
+      clientMetrics.bindTo(registry);
+      return clientMetrics;
+    }
+    return () -> {};
   }
 
   public static PemKeyCertOptions permKeyCertOptions() {
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java
index 29729c6ab..7fbc6ae06 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerMessageHandler.java
@@ -21,8 +21,10 @@ import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.eventbus.Message;
 import io.vertx.core.eventbus.MessageConsumer;
+
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -67,17 +69,19 @@ public class ResourcesReconcilerMessageHandler implements Handler<Message<Object
       logger.info("Reconciling contract {}", keyValue("contractGeneration", contract.getGeneration()));
 
       resourcesReconciler.reconcile(contract.getResourcesList())
-        .onSuccess(v -> logger.info(
-          "Reconciled contract generation {}",
-          keyValue("contractGeneration", contract.getGeneration()))
-        )
-        .onFailure(cause -> logger.error(
-          "Failed to reconcile contract generation {}",
-          keyValue("contractGeneration", contract.getGeneration()),
-          cause
-          )
-        )
         .onComplete(r -> {
+          if (r.succeeded()) {
+            logger.info(
+              "Reconciled contract generation {}",
+              keyValue("contractGeneration", contract.getGeneration())
+            );
+          } else {
+            logger.error(
+              "Failed to reconcile contract generation {}",
+              keyValue("contractGeneration", contract.getGeneration()),
+              r.cause()
+            );
+          }
 
           // We have reconciled the last known contract.
           reconciling.set(false);
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
index bba60df04..5a11923d5 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
@@ -45,7 +45,7 @@ class CredentialsValidator {
 
     final var SASLMechanism = credentials.SASLMechanism();
     if (is(SecurityProtocol.SASL_PLAINTEXT, securityProtocol)) {
-      if (isInvalidSASLMechanism(SASLMechanism)) {
+      if (SASLMechanism != null && isInvalidSASLMechanism(SASLMechanism)) {
         return "Security protocol " + securityProtocol.name + ": invalid SASL mechanism, expected SCRAM-SHA-256, SCRAM-SHA-512 or PLAIN got " + SASLMechanism;
       }
       if (anyBlank(credentials.SASLUsername(), credentials.SASLPassword())) {
@@ -55,7 +55,7 @@ class CredentialsValidator {
     }
 
     if (is(SecurityProtocol.SASL_SSL, securityProtocol)) {
-      if (isInvalidSASLMechanism(SASLMechanism)) {
+      if (SASLMechanism != null && isInvalidSASLMechanism(SASLMechanism)) {
         return "Security protocol " + securityProtocol.name + ": invalid SASL mechanism, expected SCRAM-SHA-256, SCRAM-SHA-512 or PLAIN got " + SASLMechanism;
       }
       if (anyBlank(credentials.SASLUsername(), credentials.SASLPassword())) {
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
index 8e371cac4..b85d03c2d 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
@@ -57,9 +57,12 @@ public class KafkaClientsAuth {
   }
 
   private static void sasl(final BiConsumer<String, String> propertiesSetter, final Credentials credentials) {
-    final var mechanism = credentials.SASLMechanism();
-    if (mechanism == null) {
-      throw new IllegalStateException("SASL mechanism required");
+    final String mechanism;
+    final var saslMechanism = credentials.SASLMechanism();
+    if (saslMechanism == null) {
+      mechanism = "PLAIN";
+    } else {
+      mechanism = saslMechanism;
     }
     propertiesSetter.accept(SaslConfigs.SASL_MECHANISM, mechanism);
     if ("PLAIN".equals(mechanism)) {
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
index 9b39d65fc..808f26d09 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
@@ -105,7 +105,6 @@ public class RecordDispatcherImpl implements RecordDispatcher {
     Objects.requireNonNull(deadLetterSinkSender, "provide deadLetterSinkSender");
     Objects.requireNonNull(recordDispatcherListener, "provide offsetStrategy");
     Objects.requireNonNull(responseHandler, "provide sinkResponseHandler");
-    Objects.requireNonNull(meterRegistry, "provide meterRegistry");
 
     this.resourceContext = resourceContext;
     this.filter = filter;
@@ -205,10 +204,12 @@ public class RecordDispatcherImpl implements RecordDispatcher {
     // Execute filtering
     final var pass = filter.test(recordContext.getRecord().value());
 
-    Metrics
-      .eventProcessingLatency(getTags(recordContext))
-      .register(meterRegistry)
-      .record(recordContext.performLatency());
+    if (meterRegistry != null) {
+      Metrics
+        .eventProcessingLatency(getTags(recordContext))
+        .register(meterRegistry)
+        .record(recordContext.performLatency());
+    }
 
     // reset timer to start calculating the dispatch latency.
     recordContext.resetTimer();
@@ -305,17 +306,21 @@ public class RecordDispatcherImpl implements RecordDispatcher {
 
   private void incrementEventCount(@Nullable final HttpResponse<?> response,
                                    final ConsumerRecordContext recordContext) {
-    Metrics
-      .eventCount(getTags(response, recordContext))
-      .register(meterRegistry)
-      .increment();
+    if (meterRegistry != null) {
+      Metrics
+        .eventCount(getTags(response, recordContext))
+        .register(meterRegistry)
+        .increment();
+    }
   }
 
   private void incrementDiscardedRecord(final ConsumerRecordContext recordContext) {
-    Metrics
-    .discardedEventCount(getTags(recordContext))
-    .register(meterRegistry)
-    .increment();
+    if (meterRegistry != null) {
+      Metrics
+        .discardedEventCount(getTags(recordContext))
+        .register(meterRegistry)
+        .increment();
+    }
   }
 
   private void recordDispatchLatency(final HttpResponse<?> response,
@@ -323,10 +328,12 @@ public class RecordDispatcherImpl implements RecordDispatcher {
     final var latency = recordContext.performLatency();
     logger.debug("Dispatch latency {}", keyValue("latency", latency));
 
-    Metrics
-      .eventDispatchLatency(getTags(response, recordContext))
-      .register(meterRegistry)
-      .record(latency);
+    if (meterRegistry != null) {
+      Metrics
+        .eventDispatchLatency(getTags(response, recordContext))
+        .register(meterRegistry)
+        .record(latency);
+    }
   }
 
   private HttpResponse<?> getResponse(final Throwable throwable) {
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
index 8fb7968cc..c79300332 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
@@ -17,6 +17,7 @@ package dev.knative.eventing.kafka.broker.dispatcher.impl.consumer;
 
 import dev.knative.eventing.kafka.broker.core.OrderedAsyncExecutor;
 import io.cloudevents.CloudEvent;
+import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.kafka.client.common.TopicPartition;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
@@ -29,31 +30,26 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 
+import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
+
 public class OrderedConsumerVerticle extends BaseConsumerVerticle {
 
   private static final Logger logger = LoggerFactory.getLogger(OrderedConsumerVerticle.class);
-  private static final Duration POLLING_TIMEOUT = Duration.ofMillis(100);
 
-  // poll wait ms = (POLL_WAIT_RECORD_FACTOR * pendingRecords) / (partitions - POLL_WAIT_PARTITION_FACTOR * Math.sqrt(partitions))
-  // Each record takes 10 ms to be processed
-  private static final double POLL_WAIT_RECORD_FACTOR = 10;
-  // This is used to take in account how much the parallelism is effective
-  private static final double POLL_WAIT_PARTITION_FACTOR = 1 / (double) Runtime.getRuntime().availableProcessors();
-  private static final long MAX_POLL_WAIT = 10 * 1000;
-  private static final int MIN_POLL_WAIT = 100;
+  private static final long POLLING_MS = 200L;
+  private static final Duration POLLING_TIMEOUT = Duration.ofMillis(1000L);
 
   private final Map<TopicPartition, OrderedAsyncExecutor> recordDispatcherExecutors;
 
-  private int pendingRecords;
-  private boolean stopPolling;
+  private boolean closed;
+  private long pollTimer;
+  private boolean isPollInFlight;
 
   public OrderedConsumerVerticle(Initializer initializer, Set<String> topics) {
     super(initializer, topics);
-
     this.recordDispatcherExecutors = new HashMap<>();
-
-    this.pendingRecords = 0;
-    this.stopPolling = false;
+    this.closed = false;
+    this.isPollInFlight = false;
   }
 
   @Override
@@ -62,79 +58,83 @@ public class OrderedConsumerVerticle extends BaseConsumerVerticle {
     this.consumer.subscribe(this.topics)
       .onFailure(startPromise::fail)
       .onSuccess(v -> {
-        startPromise.complete();
-
-        logger.debug("Starting polling");
+        this.pollTimer = vertx.setPeriodic(POLLING_MS, x -> poll());
         this.poll();
+        startPromise.complete();
       });
   }
 
-  public void poll() {
-    if (this.stopPolling) {
+  private void poll() {
+    if (this.closed || this.isPollInFlight || !isWaitingForTasks()) {
       return;
     }
+    this.isPollInFlight = true;
+
+    logger.debug("Polling for records {}", keyValue("topics", topics));
+
     this.consumer.poll(POLLING_TIMEOUT)
+      .onSuccess(this::recordsHandler)
       .onFailure(t -> {
-        if (this.stopPolling) {
+        if (this.closed) {
           // The failure might have been caused by stopping the consumer, so we just ignore it
           return;
         }
-        this.exceptionHandler(t);
-        this.schedulePoll();
-      })
-      .onSuccess(records -> {
-        this.recordsHandler(records);
-        this.schedulePoll();
+        isPollInFlight = false;
+        exceptionHandler(t);
       });
   }
 
-  void schedulePoll() {
-    // Check if we need to poll immediately, or wait a bit for the queues to free
-    long pollWait = pollWaitMs();
-
-    if (pollWait > MIN_POLL_WAIT) {
-      vertx.setTimer(pollWaitMs(), v -> this.poll());
-    } else {
-      // Poll immediately
-      this.poll();
-    }
-  }
-
   @Override
   public void stop(Promise<Void> stopPromise) {
     // Stop the executors
-    this.stopPolling = true;
+    this.closed = true;
+    this.vertx.cancelTimer(this.pollTimer);
     this.recordDispatcherExecutors.values().forEach(OrderedAsyncExecutor::stop);
-
     // Stop the consumer
     super.stop(stopPromise);
   }
 
-  void recordsHandler(KafkaConsumerRecords<Object, CloudEvent> records) {
-    if (records == null) {
+  private void recordsHandler(KafkaConsumerRecords<Object, CloudEvent> records) {
+    if (this.closed) {
+      return;
+    }
+    isPollInFlight = false;
+
+    if (records == null || records.size() == 0) {
       return;
     }
     // Put records in queues
     // I assume the records are ordered per topic-partition
     for (int i = 0; i < records.size(); i++) {
-      this.pendingRecords++;
-      KafkaConsumerRecord<Object, CloudEvent> record = records.recordAt(i);
-      this.enqueueRecord(new TopicPartition(record.topic(), record.partition()), record);
+      final var record = records.recordAt(i);
+      final var executor = executorFor(new TopicPartition(record.topic(), record.partition()));
+      executor.offer(() -> dispatch(record));
     }
   }
 
-  void enqueueRecord(TopicPartition topicPartition, KafkaConsumerRecord<Object, CloudEvent> record) {
-    this.recordDispatcherExecutors.computeIfAbsent(topicPartition, (tp) -> new OrderedAsyncExecutor())
-      .offer(() -> this.recordDispatcher.dispatch(record).onComplete(v -> this.pendingRecords--));
+  private Future<Void> dispatch(final KafkaConsumerRecord<Object, CloudEvent> record) {
+    if (this.closed) {
+      return Future.failedFuture("Consumer verticle closed topics=" + topics);
+    }
+    return this.recordDispatcher.dispatch(record);
   }
 
-  long pollWaitMs() {
-    double partitions = this.recordDispatcherExecutors.size();
-    long computed = Math.round(
-      (POLL_WAIT_RECORD_FACTOR * pendingRecords) /
-        (partitions - (POLL_WAIT_PARTITION_FACTOR * Math.sqrt(partitions)))
-    );
-    return Math.min(computed, MAX_POLL_WAIT);
+  private synchronized OrderedAsyncExecutor executorFor(final TopicPartition topicPartition) {
+    var executor = this.recordDispatcherExecutors.get(topicPartition);
+    if (executor != null) {
+      return executor;
+    }
+    executor = new OrderedAsyncExecutor();
+    this.recordDispatcherExecutors.put(topicPartition, executor);
+    return executor;
   }
 
+  private boolean isWaitingForTasks() {
+    for (OrderedAsyncExecutor value : this.recordDispatcherExecutors.values()) {
+      if (value.isWaitingForTasks()) {
+        return true;
+      }
+    }
+    return this.recordDispatcherExecutors.size() == 0;
+  }
 }
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 95bf3e3c6..448931f07 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
@@ -201,6 +201,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
   }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
index 11c41e054..223e23053 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
@@ -95,7 +95,6 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
   private final WebClientOptions webClientOptions;
   private final Map<String, Object> producerConfigs;
   private final AuthProvider authProvider;
-  private final Counter eventsSentCounter;
 
   /**
    * All args constructor.
@@ -116,7 +115,6 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
     Objects.requireNonNull(consumerConfigs, "provide consumerConfigs");
     Objects.requireNonNull(webClientOptions, "provide webClientOptions");
     Objects.requireNonNull(producerConfigs, "provide producerConfigs");
-    Objects.requireNonNull(metricsRegistry, "provide metricsRegistry");
 
     this.consumerConfigs = consumerConfigs.entrySet()
       .stream()
@@ -128,7 +126,6 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
       .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
     this.webClientOptions = webClientOptions;
     this.authProvider = authProvider;
-    this.eventsSentCounter = metricsRegistry.counter(Metrics.HTTP_EVENTS_SENT_COUNT);
   }
 
   /**
@@ -195,7 +192,7 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
            egressSubscriberSender,
            egressDeadLetterSender,
            responseHandler,
-           new OffsetManager(vertx, consumer, eventsSentCounter::increment, commitIntervalMs),
+           new OffsetManager(vertx, consumer, (v) -> {}, commitIntervalMs),
            ConsumerTracer.create(
              ((VertxInternal) vertx).tracer(),
              new KafkaClientOptions()


Test patch:
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
index 75796af45..1e02418ee 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
@@ -21,8 +21,13 @@ import (
 	"fmt"
 	"testing"
 
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	clientgotesting "k8s.io/client-go/testing"
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	bindings "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
 
@@ -94,6 +99,244 @@ func TestReconcileKind(t *testing.T) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal with SASL with type",
+			Objects: []runtime.Object{
+				//NewSourceSASL(),
+				&sources.KafkaSource{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: SourceNamespace,
+						Name:      SourceName,
+						UID:       SourceUUID,
+					},
+					Spec: sources.KafkaSourceSpec{
+						KafkaAuthSpec: v1beta1.KafkaAuthSpec{
+							BootstrapServers: []string{SourceBootstrapServers},
+							Net: v1beta1.KafkaNetSpec{
+								SASL: bindings.KafkaSASLSpec{
+									Enable: true,
+									User: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "user",
+										},
+									},
+									Password: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "password",
+										},
+									},
+									Type: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "type",
+										},
+									},
+								},
+								TLS: bindings.KafkaTLSSpec{
+									Enable: true,
+								},
+							},
+						},
+						Topics:        SourceTopics,
+						ConsumerGroup: SourceConsumerGroup,
+						SourceSpec: duckv1.SourceSpec{
+							Sink: NewSourceSinkReference(),
+						},
+					},
+				},
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupMetaLabels(OwnerAsSourceLabel),
+					WithConsumerGroupLabels(ConsumerSourceLabel),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(&kafkainternals.Auth{
+							NetSpec: &bindings.KafkaNetSpec{
+								SASL: bindings.KafkaSASLSpec{
+									Enable: true,
+									User: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "user",
+										},
+									},
+									Password: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "password",
+										},
+									},
+									Type: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "type",
+										},
+									},
+								},
+								TLS: bindings.KafkaTLSSpec{
+									Enable: true,
+								},
+							},
+						}),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+						SourceNetSaslTls(true),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal with SASL without type",
+			Objects: []runtime.Object{
+				//NewSourceSASL(),
+				&sources.KafkaSource{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: SourceNamespace,
+						Name:      SourceName,
+						UID:       SourceUUID,
+					},
+					Spec: sources.KafkaSourceSpec{
+						KafkaAuthSpec: v1beta1.KafkaAuthSpec{
+							BootstrapServers: []string{SourceBootstrapServers},
+							Net: v1beta1.KafkaNetSpec{
+								SASL: bindings.KafkaSASLSpec{
+									Enable: true,
+									User: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "user",
+										},
+									},
+									Password: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "password",
+										},
+									},
+								},
+								TLS: bindings.KafkaTLSSpec{
+									Enable: true,
+								},
+							},
+						},
+						Topics:        SourceTopics,
+						ConsumerGroup: SourceConsumerGroup,
+						SourceSpec: duckv1.SourceSpec{
+							Sink: NewSourceSinkReference(),
+						},
+					},
+				},
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupMetaLabels(OwnerAsSourceLabel),
+					WithConsumerGroupLabels(ConsumerSourceLabel),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(&kafkainternals.Auth{
+							NetSpec: &bindings.KafkaNetSpec{
+								SASL: bindings.KafkaSASLSpec{
+									Enable: true,
+									User: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "user",
+										},
+									},
+									Password: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "password",
+										},
+									},
+								},
+								TLS: bindings.KafkaTLSSpec{
+									Enable: true,
+								},
+							},
+						}),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+						SourceNetSaslTls(false),
+					),
+				},
+			},
+		},
 		{
 			Name: "Reconciled normal - ce overrides",
 			Objects: []runtime.Object{
@@ -558,3 +801,51 @@ func StatusSourceConsumerGroupReplicas(replicas int32) KRShapedOption {
 		ks.Status.Consumers = replicas
 	}
 }
+
+func SourceNetSaslTls(withType bool) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.Spec.Net = bindings.KafkaNetSpec{
+			SASL: bindings.KafkaSASLSpec{
+				Enable: true,
+				User: bindings.SecretValueFromSource{
+					SecretKeyRef: &corev1.SecretKeySelector{
+						LocalObjectReference: corev1.LocalObjectReference{
+							Name: SecretName,
+						},
+						Key: "user",
+					},
+				},
+				Password: bindings.SecretValueFromSource{
+					SecretKeyRef: &corev1.SecretKeySelector{
+						LocalObjectReference: corev1.LocalObjectReference{
+							Name: SecretName,
+						},
+						Key: "password",
+					},
+				},
+				// Type: bindings.SecretValueFromSource{
+				// 	SecretKeyRef: &corev1.SecretKeySelector{
+				// 		LocalObjectReference: corev1.LocalObjectReference{
+				// 			Name: SecretName,
+				// 		},
+				// 		Key: "type",
+				// 	},
+				// },
+			},
+			TLS: bindings.KafkaTLSSpec{
+				Enable: true,
+			},
+		}
+		if withType {
+			ks.Spec.Net.SASL.Type = bindings.SecretValueFromSource{
+				SecretKeyRef: &corev1.SecretKeySelector{
+					LocalObjectReference: corev1.LocalObjectReference{
+						Name: SecretName,
+					},
+					Key: "type",
+				},
+			}
+		}
+	}
+}
diff --git a/control-plane/pkg/security/secret_test.go b/control-plane/pkg/security/secret_test.go
index 604702f77..04305c539 100644
--- a/control-plane/pkg/security/secret_test.go
+++ b/control-plane/pkg/security/secret_test.go
@@ -75,6 +75,24 @@ func TestSASLPlain(t *testing.T) {
 	assert.Equal(t, "my-user-password", config.Net.SASL.Password)
 }
 
+func TestSASLPlainDefaulted(t *testing.T) {
+	secret := map[string][]byte{
+		"protocol": []byte("SASL_PLAINTEXT"),
+		"user":     []byte("my-user-name"),
+		"password": []byte("my-user-password"),
+	}
+	config := sarama.NewConfig()
+
+	err := kafka.Options(config, secretData(secret))
+
+	assert.Nil(t, err)
+	assert.True(t, config.Net.SASL.Enable)
+	assert.True(t, config.Net.SASL.Handshake)
+	assert.Equal(t, sarama.SASLMechanism(sarama.SASLTypePlaintext), config.Net.SASL.Mechanism)
+	assert.Equal(t, "my-user-name", config.Net.SASL.User)
+	assert.Equal(t, "my-user-password", config.Net.SASL.Password)
+}
+
 func TestSASLPlainLSCRAM256(t *testing.T) {
 	secret := map[string][]byte{
 		"protocol":       []byte("SASL_PLAINTEXT"),
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutorTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutorTest.java
index cdea4598c..cd74587db 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutorTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/OrderedAsyncExecutorTest.java
@@ -20,6 +20,7 @@ import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.junit5.VertxExtension;
+
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
@@ -28,6 +29,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
+
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -56,17 +58,17 @@ public class OrderedAsyncExecutorTest {
 
   @ParameterizedTest(name = "with delay {0}ms and tasks {1}")
   @MethodSource("inputArgs")
-  public void shouldExecuteInOrder(final long millis, final int tasks, Vertx vertx) throws InterruptedException {
+  public void shouldExecuteInOrder(final long millis, final int tasks, final Vertx parentVertx) throws InterruptedException {
     Random random = new Random();
 
     // Deploy the verticle
     AVerticle verticle = new AVerticle();
     CountDownLatch startVerticleLatch = new CountDownLatch(1);
-    vertx.deployVerticle(verticle, v -> startVerticleLatch.countDown());
+    parentVertx.deployVerticle(verticle, v -> startVerticleLatch.countDown());
     startVerticleLatch.await();
 
     // Rewrite the vertx instance in order to make sure we run always in the same context
-    vertx = verticle.getVertx();
+    final var vertx = verticle.getVertx();
 
     CountDownLatch tasksLatch = new CountDownLatch(tasks);
     List<Integer> executed = new ArrayList<>(tasks);
@@ -74,8 +76,8 @@ public class OrderedAsyncExecutorTest {
     OrderedAsyncExecutor asyncExecutor = new OrderedAsyncExecutor();
 
     for (int i = 0; i < tasks; i++) {
-      Supplier<Future<?>> task = generateTask(vertx, random, millis, i, tasksLatch, executed);
-      vertx.runOnContext((v) -> asyncExecutor.offer(task));
+      final var n = i;
+      vertx.runOnContext((v) -> asyncExecutor.offer(generateTask(vertx, random, millis, n, tasksLatch, executed)));
     }
 
     assertThat(
@@ -152,7 +154,7 @@ public class OrderedAsyncExecutorTest {
     }
   }
 
-  private class AVerticle extends AbstractVerticle {
+  private static class AVerticle extends AbstractVerticle {
   }
 
 }
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
index 49d590128..fcfd39e9e 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
@@ -110,12 +110,19 @@ public class CredentialsValidatorTest {
     securityProtocolSaslPlaintextValid("PLAIN");
   }
 
+  @Test
+  public void securityProtocolSaslPlaintextDefauledPlainValid() {
+    securityProtocolSaslPlaintextValid(null);
+  }
+
   private static void securityProtocolSaslPlaintextValid(final String mechanism) {
 
     final var credential = mock(Credentials.class);
 
     when(credential.securityProtocol()).thenReturn(SecurityProtocol.SASL_PLAINTEXT);
-    when(credential.SASLMechanism()).thenReturn(mechanism);
+    if(mechanism != null) {
+      when(credential.SASLMechanism()).thenReturn(mechanism);
+    }
     when(credential.SASLUsername()).thenReturn("aaa");
     when(credential.SASLPassword()).thenReturn("bbb");
 
@@ -184,6 +191,18 @@ public class CredentialsValidatorTest {
     assertThat(CredentialsValidator.validate(credential)).isNull();
   }
 
+  @Test
+  public void securityProtocolSaslDefaultedPlainSslValid() {
+    final var credential = mock(Credentials.class);
+
+    when(credential.securityProtocol()).thenReturn(SecurityProtocol.SASL_SSL);
+    when(credential.caCertificates()).thenReturn("xyz");
+    when(credential.SASLUsername()).thenReturn("aaa");
+    when(credential.SASLPassword()).thenReturn("bbb");
+
+    assertThat(CredentialsValidator.validate(credential)).isNull();
+  }
+
   @Test
   public void securityProtocolSaslSslScramSha513InValid() {
     final var credential = mock(Credentials.class);
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuthTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuthTest.java
index 8b08e44f5..50ad2d433 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuthTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuthTest.java
@@ -49,13 +49,20 @@ public class KafkaClientsAuthTest {
     shouldConfigureSaslSsl(PlainLoginModule.class, "PLAIN");
   }
 
+  @Test
+  public void shouldConfigureSaslDefaultedPlainSsl() {
+    shouldConfigureSaslSsl(PlainLoginModule.class, null);
+  }
+
   private static void shouldConfigureSaslSsl(final Class<? extends LoginModule> module, final String mechanism) {
     final var props = new Properties();
 
     final var credentials = mock(Credentials.class);
     when(credentials.securityProtocol()).thenReturn(SecurityProtocol.SASL_SSL);
     when(credentials.caCertificates()).thenReturn("xyz");
-    when(credentials.SASLMechanism()).thenReturn(mechanism);
+    if(mechanism != null) {
+      when(credentials.SASLMechanism()).thenReturn(mechanism);
+    }
     when(credentials.SASLUsername()).thenReturn("aaa");
     when(credentials.SASLPassword()).thenReturn("bbb");
 
@@ -65,7 +72,12 @@ public class KafkaClientsAuthTest {
     expected.setProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_SSL.name());
     expected.setProperty(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, DefaultSslEngineFactory.PEM_TYPE);
     expected.setProperty(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, "xyz");
-    expected.setProperty(SaslConfigs.SASL_MECHANISM, mechanism);
+    if(mechanism != null) {
+      expected.setProperty(SaslConfigs.SASL_MECHANISM, mechanism);
+    } else {
+      expected.setProperty(SaslConfigs.SASL_MECHANISM, "PLAIN"); // it is defaulted
+
+    }
     expected.setProperty(
       SaslConfigs.SASL_JAAS_CONFIG,
       module.getName() + " required username=\"" + credentials.SASLUsername() + "\" password=\"" +
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 ba7a01867..9c45a7d5f 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
@@ -77,7 +77,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(),
@@ -89,7 +89,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()
     );
   }
 }
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnvTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnvTest.java
index 6c961ec90..61ca0ce2f 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnvTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverEnvTest.java
@@ -42,9 +42,9 @@ class ReceiverEnvTest {
         case ReceiverEnv.HTTPSERVER_CONFIG_FILE_PATH -> HTTPSERVER_CONFIG_FILE_PATH;
         case BaseEnv.PRODUCER_CONFIG_FILE_PATH -> PRODUCER_CONFIG_PATH;
         case BaseEnv.DATA_PLANE_CONFIG_FILE_PATH -> DATA_PLANE_CONFIG_FILE_PATH;
+        case BaseEnv.METRICS_PUBLISH_QUANTILES -> "TRUE";
         case BaseEnv.METRICS_PORT -> "9092";
         case BaseEnv.METRICS_PATH -> "/path";
-        case BaseEnv.METRICS_PUBLISH_QUANTILES -> "TRUE";
         case BaseEnv.CONFIG_TRACING_PATH -> TRACING_CONFIG_PATH;
         case BaseEnv.METRICS_JVM_ENABLED -> METRICS_JVM_ENABLED;
         case BaseEnv.WAIT_STARTUP_SECONDS -> Integer.valueOf(WAIT_STARTUP_SECONDS).toString();