Repository: knative-sandbox/eventing-kafka-broker
Pull request #308 (2020-10-19 09:45:03)

Improve error messages in kafka-consumer test image 
If the job fails, it's hard to diagnose what went wrong, this
PR adds some more logging to help diagnose errors.

## Proposed Changes

- Improve error messages in the `kafka-consumer` test image

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/308#" title="Author self-approved">pierDipi</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/master/OWNERS)~~ [pierDipi]

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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308?src=pr&el=h1) Report
> Merging [#308](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308?src=pr&el=desc) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/a91d5dc28bc9f133c56d4fd2f456342f2e4fe8f0?el=desc) will **decrease** coverage by `1.14%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308?src=pr&el=tree)

```diff
@@             Coverage Diff              @@
##             master     #308      +/-   ##
============================================
- Coverage     76.27%   75.13%   -1.15%     
  Complexity      176      176              
============================================
  Files            49       27      -22     
  Lines          1505      756     -749     
  Branches         47       47              
============================================
- Hits           1148      568     -580     
+ Misses          285      163     -122     
+ Partials         72       25      -47     
```

| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| #java-unittests | `75.13% <ø> (ø)` | `176.00 <ø> (ø)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [control-plane/pkg/reconciler/kafka/topic.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS90b3BpYy5nbw==) | | | |
| [...ntrol-plane/pkg/reconciler/broker/broker\_config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyX2NvbmZpZy5nbw==) | | | |
| [control-plane/pkg/core/config/resource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvcmVzb3VyY2UuZ28=) | | | |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | | | |
| [control-plane/pkg/reconciler/trigger/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL2NvbnRyb2xsZXIuZ28=) | | | |
| [...ntrol-plane/pkg/apis/eventing/v1alpha1/register.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9yZWdpc3Rlci5nbw==) | | | |
| [.../pkg/apis/eventing/v1alpha1/kafka\_sink\_defaults.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2RlZmF1bHRzLmdv) | | | |
| [...ane/pkg/apis/eventing/v1alpha1/kafka\_sink\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3R5cGVzLmdv) | | | |
| [control-plane/pkg/reconciler/broker/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvY29udHJvbGxlci5nbw==) | | | |
| [...kg/apis/eventing/v1alpha1/kafka\_sink\_conversion.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2NvbnZlcnNpb24uZ28=) | | | |
| ... and [12 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308?src=pr&el=footer). Last update [a91d5dc...6cdd4ad](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/308?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/control-plane/cmd/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
index 9d67bcb6d..9796a2649 100644
--- a/control-plane/cmd/kafka-controller/main.go
+++ b/control-plane/cmd/kafka-controller/main.go
@@ -36,7 +36,7 @@ const (
 
 func main() {
 
-	brokerEnv, err := config.GetEnvConfig("BROKER")
+	brokerEnv, err := config.GetEnvConfig("BROKER", broker.ValidateDefaultBackoffDelayMs)
 	if err != nil {
 		log.Fatal("cannot process environment variables with prefix BROKER", err)
 	}
diff --git a/control-plane/config/500-controller.yaml b/control-plane/config/500-controller.yaml
index d7aad7bda..9e3aa513f 100644
--- a/control-plane/config/500-controller.yaml
+++ b/control-plane/config/500-controller.yaml
@@ -82,8 +82,8 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
-            - name: BROKER_DEFAULT_BACKOFF_DELAY
-              value: "PT1S"
+            - name: BROKER_DEFAULT_BACKOFF_DELAY_MS
+              value: "1000" # 1 second
             - name: SYSTEM_NAMESPACE
               valueFrom:
                 fieldRef:
diff --git a/control-plane/pkg/config/env_config.go b/control-plane/pkg/config/env_config.go
index 95c25c283..29ab75a63 100644
--- a/control-plane/pkg/config/env_config.go
+++ b/control-plane/pkg/config/env_config.go
@@ -20,7 +20,6 @@ import (
 	"fmt"
 
 	"github.com/kelseyhightower/envconfig"
-	"github.com/rickb777/date/period"
 )
 
 type Env struct {
@@ -30,19 +29,22 @@ type Env struct {
 	IngressName                 string `required:"true" split_words:"true"`
 	SystemNamespace             string `required:"true" split_words:"true"`
 	DataPlaneConfigFormat       string `required:"true" split_words:"true"`
-	DefaultBackoffDelay         string `required:"false" split_words:"true"`
+	DefaultBackoffDelayMs       uint64 `required:"false" split_words:"true"`
 }
 
-func GetEnvConfig(prefix string) (*Env, error) {
+// ValidationOption represents a function to validate the Env configurations.
+type ValidationOption func(env Env) error
+
+func GetEnvConfig(prefix string, validations ...ValidationOption) (*Env, error) {
 	env := &Env{}
 
 	if err := envconfig.Process(prefix, env); err != nil {
 		return nil, fmt.Errorf("failed to process env config: %w", err)
 	}
 
-	if env.DefaultBackoffDelay != "" {
-		if isNotValidBackoffDelay(env.DefaultBackoffDelay) {
-			return nil, fmt.Errorf("invalid backoff delay: %s", env.DefaultBackoffDelay)
+	for _, v := range validations {
+		if err := v(*env); err != nil {
+			return nil, fmt.Errorf("env config validation failed: %w", err)
 		}
 	}
 
@@ -52,8 +54,3 @@ func GetEnvConfig(prefix string) (*Env, error) {
 func (c *Env) DataPlaneConfigMapAsString() string {
 	return fmt.Sprintf("%s/%s", c.DataPlaneConfigMapNamespace, c.DataPlaneConfigMapName)
 }
-
-func isNotValidBackoffDelay(delay string) bool {
-	_, err := period.Parse(delay)
-	return err != nil
-}
diff --git a/control-plane/pkg/contract/contract.pb.go b/control-plane/pkg/contract/contract.pb.go
index ed2355947..78f3c409c 100644
--- a/control-plane/pkg/contract/contract.pb.go
+++ b/control-plane/pkg/contract/contract.pb.go
@@ -189,14 +189,8 @@ type EgressConfig struct {
 	Retry uint32 `protobuf:"varint,2,opt,name=retry,proto3" json:"retry,omitempty"`
 	// backoffPolicy is the retry backoff policy (linear, exponential).
 	BackoffPolicy BackoffPolicy `protobuf:"varint,3,opt,name=backoffPolicy,proto3,enum=BackoffPolicy" json:"backoffPolicy,omitempty"`
-	// backoffDelay is the delay before retrying.
-	// More information on Duration format:
-	//  - https://www.iso.org/iso-8601-date-and-time-format.html
-	//  - https://en.wikipedia.org/wiki/ISO_8601
-	//
-	// For linear policy, backoff delay is backoffDelay*<numberOfRetries>.
-	// For exponential policy, backoff delay is backoffDelay*2^<numberOfRetries>.
-	BackoffDelay string `protobuf:"bytes,4,opt,name=backoffDelay,proto3" json:"backoffDelay,omitempty"`
+	// backoffDelay is the delay before retrying in milliseconds.
+	BackoffDelay uint64 `protobuf:"varint,4,opt,name=backoffDelay,proto3" json:"backoffDelay,omitempty"`
 }
 
 func (x *EgressConfig) Reset() {
@@ -252,11 +246,11 @@ func (x *EgressConfig) GetBackoffPolicy() BackoffPolicy {
 	return BackoffPolicy_Exponential
 }
 
-func (x *EgressConfig) GetBackoffDelay() string {
+func (x *EgressConfig) GetBackoffDelay() uint64 {
 	if x != nil {
 		return x.BackoffDelay
 	}
-	return ""
+	return 0
 }
 
 type Egress struct {
@@ -637,7 +631,7 @@ var file_proto_def_contract_proto_rawDesc = []byte{
 	0x63, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0e, 0x2e, 0x42, 0x61, 0x63, 0x6b, 0x6f,
 	0x66, 0x66, 0x50, 0x6f, 0x6c, 0x69, 0x63, 0x79, 0x52, 0x0d, 0x62, 0x61, 0x63, 0x6b, 0x6f, 0x66,
 	0x66, 0x50, 0x6f, 0x6c, 0x69, 0x63, 0x79, 0x12, 0x22, 0x0a, 0x0c, 0x62, 0x61, 0x63, 0x6b, 0x6f,
-	0x66, 0x66, 0x44, 0x65, 0x6c, 0x61, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x62,
+	0x66, 0x66, 0x44, 0x65, 0x6c, 0x61, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0c, 0x62,
 	0x61, 0x63, 0x6b, 0x6f, 0x66, 0x66, 0x44, 0x65, 0x6c, 0x61, 0x79, 0x22, 0xee, 0x01, 0x0a, 0x06,
 	0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x24, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6d,
 	0x65, 0x72, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x63,
diff --git a/control-plane/pkg/core/config/utils.go b/control-plane/pkg/core/config/utils.go
index 25e9c65bf..f20e3e491 100644
--- a/control-plane/pkg/core/config/utils.go
+++ b/control-plane/pkg/core/config/utils.go
@@ -18,7 +18,9 @@ package config
 
 import (
 	"fmt"
+	"math"
 
+	"github.com/rickb777/date/period"
 	duck "knative.dev/eventing/pkg/apis/duck/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
@@ -61,13 +63,20 @@ func BackoffPolicyFromString(backoffPolicy *duck.BackoffPolicyType) contract.Bac
 	}
 }
 
-// BackoffDelayFromString returns the BackoffDelay from the given string.
+// BackoffDelayFromISO8601String returns the BackoffDelay from the given string.
 //
 // Default value is the specified defaultDelay.
-func BackoffDelayFromString(backoffDelay *string, defaultDelay string) string {
+func BackoffDelayFromISO8601String(backoffDelay *string, defaultDelay uint64) (uint64, error) {
 	if backoffDelay == nil {
-		return defaultDelay
+		return defaultDelay, nil
 	}
 
-	return *backoffDelay
+	d, err := period.Parse(*backoffDelay, false)
+	if err != nil {
+		return 0, fmt.Errorf("failed to parse backoffDelay: %w", err)
+	}
+
+	ms, _ := d.Duration()
+
+	return uint64(math.Abs(float64(ms.Milliseconds()))), nil
 }
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 61611f30c..b32f26042 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -334,7 +334,12 @@ func (r *Reconciler) getBrokerResource(ctx context.Context, topic string, broker
 		if delivery.Retry != nil {
 			ensureEgressConfig(res)
 			res.EgressConfig.Retry = uint32(*delivery.Retry)
-			res.EgressConfig.BackoffDelay = coreconfig.BackoffDelayFromString(delivery.BackoffDelay, r.Configs.DefaultBackoffDelay)
+			var err error
+			delay, err := coreconfig.BackoffDelayFromISO8601String(delivery.BackoffDelay, r.Configs.DefaultBackoffDelayMs)
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse backoff delay: %w", err)
+			}
+			res.EgressConfig.BackoffDelay = delay
 			res.EgressConfig.BackoffPolicy = coreconfig.BackoffPolicyFromString(delivery.BackoffPolicy)
 		}
 	}
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index b5479a735..af6a5299e 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -18,6 +18,7 @@ package broker
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/Shopify/sarama"
@@ -36,6 +37,7 @@ import (
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 )
@@ -113,3 +115,10 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 
 	return impl
 }
+
+func ValidateDefaultBackoffDelayMs(env config.Env) error {
+	if env.DefaultBackoffDelayMs == 0 {
+		return errors.New("default backoff delay cannot be 0")
+	}
+	return nil
+}
diff --git a/data-plane/config/sink/template/500-receiver.yaml b/data-plane/config/sink/template/500-receiver.yaml
index 3636e7074..140746347 100644
--- a/data-plane/config/sink/template/500-receiver.yaml
+++ b/data-plane/config/sink/template/500-receiver.yaml
@@ -36,7 +36,7 @@ spec:
     spec:
       securityContext:
         runAsNonRoot: true
-        runAsUser: 999
+        runAsUser: 65532
       containers:
         - name: kafka-sink-receiver
           image: ${KNATIVE_KAFKA_SINK_RECEIVER_IMAGE}
diff --git a/data-plane/config/template/500-dispatcher.yaml b/data-plane/config/template/500-dispatcher.yaml
index cbbb6eba1..7c0bc003e 100644
--- a/data-plane/config/template/500-dispatcher.yaml
+++ b/data-plane/config/template/500-dispatcher.yaml
@@ -36,7 +36,7 @@ spec:
     spec:
       securityContext:
         runAsNonRoot: true
-        runAsUser: 999
+        runAsUser: 65532
       containers:
         - name: kafka-broker-dispatcher
           image: ${KNATIVE_KAFKA_BROKER_DISPATCHER_IMAGE}
diff --git a/data-plane/config/template/500-receiver.yaml b/data-plane/config/template/500-receiver.yaml
index 0a62f6142..cc0242d9c 100644
--- a/data-plane/config/template/500-receiver.yaml
+++ b/data-plane/config/template/500-receiver.yaml
@@ -36,7 +36,7 @@ spec:
     spec:
       securityContext:
         runAsNonRoot: true
-        runAsUser: 999
+        runAsUser: 65532
       containers:
         - name: kafka-broker-receiver
           image: ${KNATIVE_KAFKA_BROKER_RECEIVER_IMAGE}
diff --git a/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java b/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
index c18ae5005..893c952cb 100644
--- a/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
+++ b/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
@@ -1177,33 +1177,13 @@ public final class DataPlaneContract {
 
     /**
      * <pre>
-     * backoffDelay is the delay before retrying.
-     * More information on Duration format:
-     *  - https://www.iso.org/iso-8601-date-and-time-format.html
-     *  - https://en.wikipedia.org/wiki/ISO_8601
-     * For linear policy, backoff delay is backoffDelay*&lt;numberOfRetries&gt;.
-     * For exponential policy, backoff delay is backoffDelay*2^&lt;numberOfRetries&gt;.
+     * backoffDelay is the delay before retrying in milliseconds.
      * </pre>
      *
-     * <code>string backoffDelay = 4;</code>
+     * <code>uint64 backoffDelay = 4;</code>
      * @return The backoffDelay.
      */
-    java.lang.String getBackoffDelay();
-    /**
-     * <pre>
-     * backoffDelay is the delay before retrying.
-     * More information on Duration format:
-     *  - https://www.iso.org/iso-8601-date-and-time-format.html
-     *  - https://en.wikipedia.org/wiki/ISO_8601
-     * For linear policy, backoff delay is backoffDelay*&lt;numberOfRetries&gt;.
-     * For exponential policy, backoff delay is backoffDelay*2^&lt;numberOfRetries&gt;.
-     * </pre>
-     *
-     * <code>string backoffDelay = 4;</code>
-     * @return The bytes for backoffDelay.
-     */
-    com.google.protobuf.ByteString
-        getBackoffDelayBytes();
+    long getBackoffDelay();
   }
   /**
    * Protobuf type {@code EgressConfig}
@@ -1220,7 +1200,6 @@ public final class DataPlaneContract {
     private EgressConfig() {
       deadLetter_ = "";
       backoffPolicy_ = 0;
-      backoffDelay_ = "";
     }
 
     @java.lang.Override
@@ -1270,10 +1249,9 @@ public final class DataPlaneContract {
               backoffPolicy_ = rawValue;
               break;
             }
-            case 34: {
-              java.lang.String s = input.readStringRequireUtf8();
+            case 32: {
 
-              backoffDelay_ = s;
+              backoffDelay_ = input.readUInt64();
               break;
             }
             default: {
@@ -1399,59 +1377,18 @@ public final class DataPlaneContract {
     }
 
     public static final int BACKOFFDELAY_FIELD_NUMBER = 4;
-    private volatile java.lang.Object backoffDelay_;
+    private long backoffDelay_;
     /**
      * <pre>
-     * backoffDelay is the delay before retrying.
-     * More information on Duration format:
-     *  - https://www.iso.org/iso-8601-date-and-time-format.html
-     *  - https://en.wikipedia.org/wiki/ISO_8601
-     * For linear policy, backoff delay is backoffDelay*&lt;numberOfRetries&gt;.
-     * For exponential policy, backoff delay is backoffDelay*2^&lt;numberOfRetries&gt;.
+     * backoffDelay is the delay before retrying in milliseconds.
      * </pre>
      *
-     * <code>string backoffDelay = 4;</code>
+     * <code>uint64 backoffDelay = 4;</code>
      * @return The backoffDelay.
      */
     @java.lang.Override
-    public java.lang.String getBackoffDelay() {
-      java.lang.Object ref = backoffDelay_;
-      if (ref instanceof java.lang.String) {
-        return (java.lang.String) ref;
-      } else {
-        com.google.protobuf.ByteString bs = 
-            (com.google.protobuf.ByteString) ref;
-        java.lang.String s = bs.toStringUtf8();
-        backoffDelay_ = s;
-        return s;
-      }
-    }
-    /**
-     * <pre>
-     * backoffDelay is the delay before retrying.
-     * More information on Duration format:
-     *  - https://www.iso.org/iso-8601-date-and-time-format.html
-     *  - https://en.wikipedia.org/wiki/ISO_8601
-     * For linear policy, backoff delay is backoffDelay*&lt;numberOfRetries&gt;.
-     * For exponential policy, backoff delay is backoffDelay*2^&lt;numberOfRetries&gt;.
-     * </pre>
-     *
-     * <code>string backoffDelay = 4;</code>
-     * @return The bytes for backoffDelay.
-     */
-    @java.lang.Override
-    public com.google.protobuf.ByteString
-        getBackoffDelayBytes() {
-      java.lang.Object ref = backoffDelay_;
-      if (ref instanceof java.lang.String) {
-        com.google.protobuf.ByteString b = 
-            com.google.protobuf.ByteString.copyFromUtf8(
-                (java.lang.String) ref);
-        backoffDelay_ = b;
-        return b;
-      } else {
-        return (com.google.protobuf.ByteString) ref;
-      }
+    public long getBackoffDelay() {
+      return backoffDelay_;
     }
 
     private byte memoizedIsInitialized = -1;
@@ -1477,8 +1414,8 @@ public final class DataPlaneContract {
       if (backoffPolicy_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.BackoffPolicy.Exponential.getNumber()) {
         output.writeEnum(3, backoffPolicy_);
       }
-      if (!getBackoffDelayBytes().isEmpty()) {
-        com.google.protobuf.GeneratedMessageV3.writeString(output, 4, backoffDelay_);
+      if (backoffDelay_ != 0L) {
+        output.writeUInt64(4, backoffDelay_);
       }
       unknownFields.writeTo(output);
     }
@@ -1500,8 +1437,9 @@ public final class DataPlaneContract {
         size += com.google.protobuf.CodedOutputStream
           .computeEnumSize(3, backoffPolicy_);
       }
-      if (!getBackoffDelayBytes().isEmpty()) {
-        size += com.google.protobuf.GeneratedMessageV3.computeStringSize(4, backoffDelay_);
+      if (backoffDelay_ != 0L) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeUInt64Size(4, backoffDelay_);
       }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
@@ -1523,8 +1461,8 @@ public final class DataPlaneContract {
       if (getRetry()
           != other.getRetry()) return false;
       if (backoffPolicy_ != other.backoffPolicy_) return false;
-      if (!getBackoffDelay()
-          .equals(other.getBackoffDelay())) return false;
+      if (getBackoffDelay()
+          != other.getBackoffDelay()) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -1543,7 +1481,8 @@ public final class DataPlaneContract {
       hash = (37 * hash) + BACKOFFPOLICY_FIELD_NUMBER;
       hash = (53 * hash) + backoffPolicy_;
       hash = (37 * hash) + BACKOFFDELAY_FIELD_NUMBER;
-      hash = (53 * hash) + getBackoffDelay().hashCode();
+      hash = (53 * hash) + com.google.protobuf.Internal.hashLong(
+          getBackoffDelay());
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -1683,7 +1622,7 @@ public final class DataPlaneContract {
 
         backoffPolicy_ = 0;
 
-        backoffDelay_ = "";
+        backoffDelay_ = 0L;
 
         return this;
       }
@@ -1773,9 +1712,8 @@ public final class DataPlaneContract {
         if (other.backoffPolicy_ != 0) {
           setBackoffPolicyValue(other.getBackoffPolicyValue());
         }
-        if (!other.getBackoffDelay().isEmpty()) {
-          backoffDelay_ = other.backoffDelay_;
-          onChanged();
+        if (other.getBackoffDelay() != 0L) {
+          setBackoffDelay(other.getBackoffDelay());
         }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
@@ -2025,123 +1963,45 @@ public final class DataPlaneContract {
         return this;
       }
 
-      private java.lang.Object backoffDelay_ = "";
+      private long backoffDelay_ ;
       /**
        * <pre>
-       * backoffDelay is the delay before retrying.
-       * More information on Duration format:
-       *  - https://www.iso.org/iso-8601-date-and-time-format.html
-       *  - https://en.wikipedia.org/wiki/ISO_8601
-       * For linear policy, backoff delay is backoffDelay*&lt;numberOfRetries&gt;.
-       * For exponential policy, backoff delay is backoffDelay*2^&lt;numberOfRetries&gt;.
+       * backoffDelay is the delay before retrying in milliseconds.
        * </pre>
        *
-       * <code>string backoffDelay = 4;</code>
+       * <code>uint64 backoffDelay = 4;</code>
        * @return The backoffDelay.
        */
-      public java.lang.String getBackoffDelay() {
-        java.lang.Object ref = backoffDelay_;
-        if (!(ref instanceof java.lang.String)) {
-          com.google.protobuf.ByteString bs =
-              (com.google.protobuf.ByteString) ref;
-          java.lang.String s = bs.toStringUtf8();
-          backoffDelay_ = s;
-          return s;
-        } else {
-          return (java.lang.String) ref;
-        }
-      }
-      /**
-       * <pre>
-       * backoffDelay is the delay before retrying.
-       * More information on Duration format:
-       *  - https://www.iso.org/iso-8601-date-and-time-format.html
-       *  - https://en.wikipedia.org/wiki/ISO_8601
-       * For linear policy, backoff delay is backoffDelay*&lt;numberOfRetries&gt;.
-       * For exponential policy, backoff delay is backoffDelay*2^&lt;numberOfRetries&gt;.
-       * </pre>
-       *
-       * <code>string backoffDelay = 4;</code>
-       * @return The bytes for backoffDelay.
-       */
-      public com.google.protobuf.ByteString
-          getBackoffDelayBytes() {
-        java.lang.Object ref = backoffDelay_;
-        if (ref instanceof String) {
-          com.google.protobuf.ByteString b = 
-              com.google.protobuf.ByteString.copyFromUtf8(
-                  (java.lang.String) ref);
-          backoffDelay_ = b;
-          return b;
-        } else {
-          return (com.google.protobuf.ByteString) ref;
-        }
+      @java.lang.Override
+      public long getBackoffDelay() {
+        return backoffDelay_;
       }
       /**
        * <pre>
-       * backoffDelay is the delay before retrying.
-       * More information on Duration format:
-       *  - https://www.iso.org/iso-8601-date-and-time-format.html
-       *  - https://en.wikipedia.org/wiki/ISO_8601
-       * For linear policy, backoff delay is backoffDelay*&lt;numberOfRetries&gt;.
-       * For exponential policy, backoff delay is backoffDelay*2^&lt;numberOfRetries&gt;.
+       * backoffDelay is the delay before retrying in milliseconds.
        * </pre>
        *
-       * <code>string backoffDelay = 4;</code>
+       * <code>uint64 backoffDelay = 4;</code>
        * @param value The backoffDelay to set.
        * @return This builder for chaining.
        */
-      public Builder setBackoffDelay(
-          java.lang.String value) {
-        if (value == null) {
-    throw new NullPointerException();
-  }
-  
+      public Builder setBackoffDelay(long value) {
+        
         backoffDelay_ = value;
         onChanged();
         return this;
       }
       /**
        * <pre>
-       * backoffDelay is the delay before retrying.
-       * More information on Duration format:
-       *  - https://www.iso.org/iso-8601-date-and-time-format.html
-       *  - https://en.wikipedia.org/wiki/ISO_8601
-       * For linear policy, backoff delay is backoffDelay*&lt;numberOfRetries&gt;.
-       * For exponential policy, backoff delay is backoffDelay*2^&lt;numberOfRetries&gt;.
+       * backoffDelay is the delay before retrying in milliseconds.
        * </pre>
        *
-       * <code>string backoffDelay = 4;</code>
+       * <code>uint64 backoffDelay = 4;</code>
        * @return This builder for chaining.
        */
       public Builder clearBackoffDelay() {
         
-        backoffDelay_ = getDefaultInstance().getBackoffDelay();
-        onChanged();
-        return this;
-      }
-      /**
-       * <pre>
-       * backoffDelay is the delay before retrying.
-       * More information on Duration format:
-       *  - https://www.iso.org/iso-8601-date-and-time-format.html
-       *  - https://en.wikipedia.org/wiki/ISO_8601
-       * For linear policy, backoff delay is backoffDelay*&lt;numberOfRetries&gt;.
-       * For exponential policy, backoff delay is backoffDelay*2^&lt;numberOfRetries&gt;.
-       * </pre>
-       *
-       * <code>string backoffDelay = 4;</code>
-       * @param value The bytes for backoffDelay to set.
-       * @return This builder for chaining.
-       */
-      public Builder setBackoffDelayBytes(
-          com.google.protobuf.ByteString value) {
-        if (value == null) {
-    throw new NullPointerException();
-  }
-  checkByteStringIsUtf8(value);
-        
-        backoffDelay_ = value;
+        backoffDelay_ = 0L;
         onChanged();
         return this;
       }
@@ -7892,7 +7752,7 @@ public final class DataPlaneContract {
       "utesEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028" +
       "\001\"n\n\014EgressConfig\022\022\n\ndeadLetter\030\001 \001(\t\022\r\n" +
       "\005retry\030\002 \001(\r\022%\n\rbackoffPolicy\030\003 \001(\0162\016.Ba" +
-      "ckoffPolicy\022\024\n\014backoffDelay\030\004 \001(\t\"\252\001\n\006Eg" +
+      "ckoffPolicy\022\024\n\014backoffDelay\030\004 \001(\004\"\252\001\n\006Eg" +
       "ress\022\025\n\rconsumerGroup\030\001 \001(\t\022\023\n\013destinati" +
       "on\030\002 \001(\t\022\022\n\010replyUrl\030\003 \001(\tH\000\0226\n\024replyToO" +
       "riginalTopic\030\004 \001(\0132\026.google.protobuf.Emp" +
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java
index efb3deb32..9cc232297 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java
@@ -37,7 +37,6 @@ import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.ext.web.client.HttpResponse;
 import io.vertx.ext.web.client.WebClient;
-import java.time.Duration;
 import java.util.Objects;
 import java.util.Properties;
 import java.util.function.Function;
@@ -47,13 +46,9 @@ import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class HttpConsumerVerticleFactory implements ConsumerVerticleFactory {
 
-  private static final Logger logger = LoggerFactory.getLogger(HttpConsumerVerticleFactory.class);
-
   private final static ConsumerRecordSender<String, CloudEvent, HttpResponse<Buffer>> NO_DLQ_SENDER =
     record -> Future.failedFuture("no DLQ set");
 
@@ -206,21 +201,16 @@ public class HttpConsumerVerticleFactory implements ConsumerVerticleFactory {
     );
   }
 
-  private static Function<Integer, Long> computeRetryPolicy(EgressConfig egress) {
-    if (egress != null && egress.getBackoffPolicy() != null) {
-      try {
-        final var delay = Duration.parse(egress.getBackoffDelay()).toMillis();
-
-        return switch (egress.getBackoffPolicy()) {
-          case Linear -> retryCount -> linearRetryPolicy(retryCount, delay);
-          case Exponential, UNRECOGNIZED -> retryCount -> exponentialRetryPolicy(retryCount, delay);
-        };
-
-      } catch (final Exception ex) {
-        logger.error("failed to set retry policy", ex);
-      }
+  /* package visibility for test */
+  static Function<Integer, Long> computeRetryPolicy(final EgressConfig egress) {
+    if (egress != null && egress.getBackoffPolicy() != null && egress.getBackoffDelay() > 0) {
+      final var delay = egress.getBackoffDelay();
+      return switch (egress.getBackoffPolicy()) {
+        case Linear -> retryCount -> linearRetryPolicy(retryCount, delay);
+        case Exponential, UNRECOGNIZED -> retryCount -> exponentialRetryPolicy(retryCount, delay);
+      };
     }
-    return retry -> 0L; // Default Vert.x retry policy
+    return retry -> 0L; // Default Vert.x retry policy, it means don't retry
   }
 
   private static Long exponentialRetryPolicy(final int retryCount, final long delay) {
diff --git a/data-plane/docker/Dockerfile b/data-plane/docker/Dockerfile
index 202fa7255..782c0e0fb 100644
--- a/data-plane/docker/Dockerfile
+++ b/data-plane/docker/Dockerfile
@@ -20,7 +20,7 @@ FROM ${JAVA_IMAGE} as builder
 ARG APP_DIR
 ARG APP_JAR
 
-WORKDIR /app
+WORKDIR /build
 
 # https://github.com/AdoptOpenJDK/openjdk-docker/issues/260
 RUN DEBIAN_FRONTEND=noninteractive apt-get update && apt-get install -y binutils
@@ -42,24 +42,13 @@ COPY . .
 
 RUN ./mvnw package -pl=${APP_DIR} -am -DskipTests -Deditorconfig.skip --no-transfer-progress
 
-RUN ./generate_jdk.sh /app/${APP_DIR}/target/${APP_JAR}
+RUN ./generate_jdk.sh /build/${APP_DIR}/target/${APP_JAR}
 
-FROM ${BASE_IMAGE} as running
+RUN cp /build/${APP_DIR}/target/${APP_JAR} /app/app.jar
 
-ARG APP_DIR
-ARG APP_JAR
-
-# Create appuser and directories
-RUN groupadd -g 999 appuser && useradd -r -u 999 -g appuser appuser && \
-  mkdir /tmp/vertx-cache && chown -R appuser:appuser /tmp/vertx-cache && \
-  mkdir /app
+FROM ${BASE_IMAGE} as running
 
-# Copy jar and jdk inside /app
-COPY --from=builder /app/jdk /app/jdk
-COPY --from=builder /app/${APP_DIR}/target/${APP_JAR} /app/app.jar
-RUN chown -R appuser:appuser /app
+COPY --from=builder /app /app
 
-# Set appuser and configure PATH
-USER appuser
-WORKDIR /app
+# Configure PATH
 ENV PATH="/app/jdk/bin:${PATH}"
diff --git a/go.mod b/go.mod
index bc13ece5c..8378a157a 100644
--- a/go.mod
+++ b/go.mod
@@ -11,12 +11,11 @@ require (
 	github.com/google/uuid v1.1.2
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/pierdipi/sacura v0.0.0-20201003135834-e90e3a725ff9
-	github.com/rickb777/date v1.13.0
+	github.com/rickb777/date v1.14.1
 	github.com/stretchr/testify v1.6.1
 	go.opencensus.io v0.22.5 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
 	go.uber.org/zap v1.16.0
-	golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0 // indirect
 	google.golang.org/protobuf v1.25.0
 	k8s.io/api v0.18.8
 	k8s.io/apiextensions-apiserver v0.18.4
diff --git a/go.sum b/go.sum
index 3ceca8494..116cb25c1 100644
--- a/go.sum
+++ b/go.sum
@@ -363,6 +363,7 @@ github.com/elazarl/goproxy v0.0.0-20180725130230-947c36da3153/go.mod h1:/Zj4wYkg
 github.com/emicklei/go-restful v0.0.0-20170410110728-ff4f55a20633/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
 github.com/emicklei/go-restful v2.9.5+incompatible h1:spTtZBk5DYEvbxMVutUuTyh1Ao2r4iyvLdACqsl/Ljk=
 github.com/emicklei/go-restful v2.9.5+incompatible/go.mod h1:otzb+WCGbkyDHkqmQmT5YD2WR4BBwUdeQoFo8l/7tVs=
+github.com/emirpasic/gods v1.12.0 h1:QAUIPSaCu4G+POclxeqb3F+WPpdKqFGlw36+yOzGlrg=
 github.com/emirpasic/gods v1.12.0/go.mod h1:YfzfFFoVP/catgzJb4IKIqXjX78Ha8FMSDh3ymbK86o=
 github.com/envoyproxy/go-control-plane v0.6.9/go.mod h1:SBwIajubJHhxtWwsL9s8ss4safvEdbitLhGGK48rN6g=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
@@ -618,6 +619,7 @@ github.com/google/gofuzz v1.1.0 h1:Hsa8mG0dQ46ij8Sl2AYJDUv1oA9/d6Vk+3LG99Oe02g=
 github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/licenseclassifier v0.0.0-20190926221455-842c0d70d702/go.mod h1:qsqn2hxC+vURpyBRygGUuinTO42MFRLcsmQ/P8v94+M=
 github.com/google/licenseclassifier v0.0.0-20200402202327-879cb1424de0/go.mod h1:qsqn2hxC+vURpyBRygGUuinTO42MFRLcsmQ/P8v94+M=
+github.com/google/licenseclassifier v0.0.0-20200708223521-3d09a0ea2f39 h1:nkr7S2ETn5pAuBeeoZggV5nXSwOm4nBLz3vscQfA/A8=
 github.com/google/licenseclassifier v0.0.0-20200708223521-3d09a0ea2f39/go.mod h1:qsqn2hxC+vURpyBRygGUuinTO42MFRLcsmQ/P8v94+M=
 github.com/google/mako v0.0.0-20190821191249-122f8dcef9e3 h1:/o5e44nTD/QEEiWPGSFT3bSqcq3Qg7q27N9bv4gKh5M=
 github.com/google/mako v0.0.0-20190821191249-122f8dcef9e3/go.mod h1:YzLcVlL+NqWnmUEPuhS1LxDDwGO9WNbVlEXaF4IH35g=
@@ -744,6 +746,7 @@ github.com/imdario/mergo v0.3.7/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJ
 github.com/imdario/mergo v0.3.8/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/imdario/mergo v0.3.9 h1:UauaLniWCFHWd+Jp9oCEkTBj8VO/9DKg3PV3VCNMDIg=
 github.com/imdario/mergo v0.3.9/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
+github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/influxdata/influxdb v0.0.0-20161215172503-049f9b42e9a5/go.mod h1:qZna6X/4elxqT3yI9iZYdZrWWdeFOOprn86kgg4+IzY=
 github.com/influxdata/tdigest v0.0.0-20180711151920-a7d76c6f093a/go.mod h1:9GkyshztGufsdPQWjH+ifgnIr3xNUL5syI70g2dzU1o=
@@ -752,6 +755,7 @@ github.com/influxdata/tdigest v0.0.0-20191024211133-5d87a7585faa/go.mod h1:Z0kXn
 github.com/influxdata/tdigest v0.0.1 h1:XpFptwYmnEKUqmkcDjrzffswZ3nvNeevbUSLPP/ZzIY=
 github.com/influxdata/tdigest v0.0.1/go.mod h1:Z0kXnxzbTC2qrx4NaIzYkE1k66+6oEDQTvL95hQFh5Y=
 github.com/jarcoal/httpmock v1.0.5/go.mod h1:ATjnClrvW/3tijVmpL/va5Z3aAyGvqU3gCT8nX0Txik=
+github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOlocH6Fxy8MmwDt+yVQYULKfN0RoTN8A=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo=
 github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o=
 github.com/jcmturner/gofork v1.0.0 h1:J7uCkflzTEhUZ64xqKnkDxq3kzc96ajM1Gli5ktUem8=
@@ -800,6 +804,7 @@ github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:C
 github.com/kelseyhightower/envconfig v1.3.0/go.mod h1:cccZRl6mQpaq41TPp5QxidR+Sa3axMbJDNb//FQX6Gg=
 github.com/kelseyhightower/envconfig v1.4.0 h1:Im6hONhd3pLkfDFsbRgu68RDNkGF1r3dvMUtDTo2cv8=
 github.com/kelseyhightower/envconfig v1.4.0/go.mod h1:cccZRl6mQpaq41TPp5QxidR+Sa3axMbJDNb//FQX6Gg=
+github.com/kevinburke/ssh_config v0.0.0-20190725054713-01f96b0aa0cd h1:Coekwdh0v2wtGp9Gmz1Ze3eVRAWJMLokvN3QjdzCHLY=
 github.com/kevinburke/ssh_config v0.0.0-20190725054713-01f96b0aa0cd/go.mod h1:CT57kijsi8u/K/BOFA39wgDQJ9CxiF4nAY/ojJ6r6mM=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
@@ -893,6 +898,7 @@ github.com/miekg/dns v1.1.17/go.mod h1:WgzbA6oji13JREwiNsRDNfl7jYdPnmz+VEuLrA+/4
 github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=
 github.com/mitchellh/copystructure v1.0.0/go.mod h1:SNtv71yrdKgLRyLFxmLdkAbkKEFWgYaq1OVrnRcwhnw=
 github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
+github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/go-ps v0.0.0-20190716172923-621e5597135b/go.mod h1:r1VsdOzOPt1ZSrGZWFoNhsAedKnEd6r9Np1+5blZCWk=
 github.com/mitchellh/go-testing-interface v1.0.0/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI=
@@ -964,6 +970,8 @@ github.com/onsi/gomega v1.8.1/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoT
 github.com/onsi/gomega v1.9.0/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoTdcA=
 github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE=
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
+github.com/onsi/gomega v1.10.2 h1:aY/nuoWlKJud2J6U0E3NWsjlg+0GtwXxgEqthRdzlcs=
+github.com/onsi/gomega v1.10.2/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/op/go-logging v0.0.0-20160315200505-970db520ece7/go.mod h1:HzydrMdWErDVzsI23lYNej1Htcns9BCg93Dk0bBINWk=
 github.com/opencontainers/go-digest v0.0.0-20170106003457-a6d0ee40d420/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
 github.com/opencontainers/go-digest v0.0.0-20180430190053-c9281466c8b2/go.mod h1:cMLVZDEM3+U2I4VmLI6N8jQYUd2OVphdqWwCJHrFt2s=
@@ -1073,8 +1081,12 @@ github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0/go.mod h1:bCqn
 github.com/remyoudompheng/bigfft v0.0.0-20170806203942-52369c62f446/go.mod h1:uYEyJGbgTkfkS4+E/PavXkNJcbFIpEtjt2B0KDQ5+9M=
 github.com/rickb777/date v1.13.0 h1:+8AmwLuY1d/rldzdqvqTEg7107bZ8clW37x4nsdG3Hs=
 github.com/rickb777/date v1.13.0/go.mod h1:GZf3LoGnxPWjX+/1TXOuzHefZFDovTyNLHDMd3qH70k=
+github.com/rickb777/date v1.14.1 h1:1WaFb5bvotNQl97xzeerUAkYi0PipNrY8xIcYSIQERI=
+github.com/rickb777/date v1.14.1/go.mod h1:swmf05C+hN+m8/Xh7gEq3uB6QJDNc5pQBWojKdHetOs=
 github.com/rickb777/plural v1.2.1 h1:UitRAgR70+yHFt26Tmj/F9dU9aV6UfjGXSbO1DcC9/U=
 github.com/rickb777/plural v1.2.1/go.mod h1:j058+3M5QQFgcZZ2oKIOekcygoZUL8gKW5yRO14BuAw=
+github.com/rickb777/plural v1.2.2 h1:4CU5NiUqXSM++2+7JCrX+oguXd2D7RY5O1YisMw1yCI=
+github.com/rickb777/plural v1.2.2/go.mod h1:xyHbelv4YvJE51gjMnHvk+U2e9zIysg6lTnSQK8XUYA=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
@@ -1101,6 +1113,7 @@ github.com/securego/gosec v0.0.0-20200103095621-79fbf3af8d83/go.mod h1:vvbZ2Ae7A
 github.com/securego/gosec v0.0.0-20200401082031-e946c8c39989/go.mod h1:i9l/TNj+yDFh9SZXUTvspXTjbFXgZGP/UvhU1S65A4A=
 github.com/securego/gosec/v2 v2.3.0/go.mod h1:UzeVyUXbxukhLeHKV3VVqo7HdoQR9MrRfFmZYotn8ME=
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
+github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
 github.com/shirou/gopsutil v0.0.0-20190901111213-e4ec7b275ada/go.mod h1:WWnYX4lzhCH5h/3YBfyVA3VbLYjlMZZAQcW9ojMexNc=
 github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4/go.mod h1:qsXQc7+bwAM3Q1u/4XEfrquwF8Lw7D7y5cD8CuHnfIc=
@@ -1139,6 +1152,7 @@ github.com/spf13/cobra v0.0.2-0.20171109065643-2da4a54c5cee/go.mod h1:1l0Ry5zgKv
 github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
 github.com/spf13/cobra v0.0.5/go.mod h1:3K3wKZymM7VvHMDS9+Akkh4K60UwM26emMESw8tLCHU=
 github.com/spf13/cobra v0.0.6/go.mod h1:/6GTrnGXV9HjY+aR4k0oJ5tcvakLuG6EuKReYlHNrgE=
+github.com/spf13/cobra v1.0.0 h1:6m/oheQuQ13N9ks4hubMG6BnvwOeaJrqSPLahSnczz8=
 github.com/spf13/cobra v1.0.0/go.mod h1:/6GTrnGXV9HjY+aR4k0oJ5tcvakLuG6EuKReYlHNrgE=
 github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=
 github.com/spf13/jwalterweatherman v1.1.0/go.mod h1:aNWZUN0dPAAO/Ljvb5BEdw96iTZ0EXowPYD95IqWIGo=
@@ -1155,6 +1169,7 @@ github.com/spf13/viper v1.6.1/go.mod h1:t3iDnF5Jlj76alVNuyFBk5oUMCvsrkbvZK0WQdfD
 github.com/spf13/viper v1.6.2/go.mod h1:t3iDnF5Jlj76alVNuyFBk5oUMCvsrkbvZK0WQdfDi5k=
 github.com/spf13/viper v1.7.0/go.mod h1:8WkrPz2fc9jxqZNCJI/76HCieCp4Q8HaLFoCha5qpdg=
 github.com/sqs/goreturns v0.0.0-20181028201513-538ac6014518/go.mod h1:CKI4AZ4XmGV240rTHfO0hfE83S6/a3/Q1siZJ/vXf7A=
+github.com/src-d/gcfg v1.4.0 h1:xXbNR5AlLSA315x2UO+fTSSAXCDf+Ar38/6oyGbDKQ4=
 github.com/src-d/gcfg v1.4.0/go.mod h1:p/UMsR43ujA89BJY9duynAwIpvqEujIH/jFlfL7jWoI=
 github.com/streadway/amqp v0.0.0-20190404075320-75d898a42a94/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/streadway/quantile v0.0.0-20150917103942-b0c588724d25 h1:7z3LSn867ex6VSaahyKadf4WtSsJIgne6A1WLOAGM8A=
@@ -1226,6 +1241,7 @@ github.com/vmware/govmomi v0.20.3/go.mod h1:URlwyTFZX72RmxtxuaFL2Uj3fD1JTvZdx59b
 github.com/wavesoftware/go-ensure v1.0.0/go.mod h1:K2UAFSwMTvpiRGay/M3aEYYuurcR8S4A6HkQlJPV8k4=
 github.com/xanzy/go-gitlab v0.31.0/go.mod h1:sPLojNBn68fMUWSxIJtdVVIP8uSBYqesTfDUseX11Ug=
 github.com/xanzy/go-gitlab v0.32.0/go.mod h1:sPLojNBn68fMUWSxIJtdVVIP8uSBYqesTfDUseX11Ug=
+github.com/xanzy/ssh-agent v0.2.1 h1:TCbipTQL2JiiCprBWx9frJ2eJlCYT00NmctrHxVAr70=
 github.com/xanzy/ssh-agent v0.2.1/go.mod h1:mLlQY/MoOhWBj+gOGMQkOeiEvkx+8pJSI+0Bx9h2kr4=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
 github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
@@ -1811,12 +1827,15 @@ gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24
 gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo=
 gopkg.in/robfig/cron.v2 v2.0.0-20150107220207-be2e0b0deed5/go.mod h1:hiOFpYm0ZJbusNj2ywpbrXowU3G8U6GIQzqn2mw1UIE=
 gopkg.in/square/go-jose.v2 v2.2.2/go.mod h1:M9dMgbHiYLoDGQrXy7OpJDJWiKiU//h+vD76mk0e1AI=
+gopkg.in/src-d/go-billy.v4 v4.3.2 h1:0SQA1pRztfTFx2miS8sA97XvooFeNOmvUenF4o0EcVg=
 gopkg.in/src-d/go-billy.v4 v4.3.2/go.mod h1:nDjArDMp+XMs1aFAESLRjfGSgfvoYN0hDfzEk0GjC98=
 gopkg.in/src-d/go-git-fixtures.v3 v3.5.0/go.mod h1:dLBcvytrw/TYZsNTWCnkNF2DSIlzWYqTe3rJR56Ac7g=
+gopkg.in/src-d/go-git.v4 v4.13.1 h1:SRtFyV8Kxc0UP7aCHcijOMQGPxHSmMOPrzulQWolkYE=
 gopkg.in/src-d/go-git.v4 v4.13.1/go.mod h1:nx5NYcxdKxq5fpltdHnPa2Exj4Sx0EclMWZQbYDu2z8=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/warnings.v0 v0.1.1/go.mod h1:jksf8JmL6Qr/oQM2OXTHunEvvTAsrWBLb6OOjuVWRNI=
+gopkg.in/warnings.v0 v0.1.2 h1:wFXVbFY8DY5/xOe1ECiWdKCzZlxgshcYVNkBHstARME=
 gopkg.in/warnings.v0 v0.1.2/go.mod h1:jksf8JmL6Qr/oQM2OXTHunEvvTAsrWBLb6OOjuVWRNI=
 gopkg.in/yaml.v2 v2.0.0-20170812160011-eb3733d160e7/go.mod h1:JAlM8MvJe8wmxCU4Bli9HhUf9+ttbYbLASfIpnQbh74=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
@@ -1935,7 +1954,6 @@ knative.dev/test-infra v0.0.0-20200505052144-5ea2f705bb55/go.mod h1:WqF1Azka+FxP
 knative.dev/test-infra v0.0.0-20200513011557-d03429a76034/go.mod h1:aMif0KXL4g19YCYwsy4Ocjjz5xgPlseYV+B95Oo4JGE=
 knative.dev/test-infra v0.0.0-20200519015156-82551620b0a9/go.mod h1:A5b2OAXTOeHT3hHhVQm3dmtbuWvIDP7qzgtqxA3/2pE=
 knative.dev/test-infra v0.0.0-20200707183444-aed09e56ddc7/go.mod h1:RjYAhXnZqeHw9+B0zsbqSPlae0lCvjekO/nw5ZMpLCs=
-knative.dev/test-infra v0.0.0-20201013100530-45e0761df397 h1:4TDu2+B8Rr6Wdcb5++tmj2UVx0mnh+DCP5Rg/iblTBk=
 knative.dev/test-infra v0.0.0-20201013100530-45e0761df397/go.mod h1:PwM5WCaFn0HlVlosnbH0dE+IzvwXbuLZ/2jCQzDX3i0=
 knative.dev/test-infra v0.0.0-20201014021030-ae3984a33f82 h1:/HZKnJdqoGjknSM50+SLHDASF6wVDYOOEXsXXLzD1QU=
 knative.dev/test-infra v0.0.0-20201014021030-ae3984a33f82/go.mod h1:lm0U0k6ARXr8Po0LcSTCfETg63LOF3ajUxcoqy5JSH0=
diff --git a/proto/def/contract.proto b/proto/def/contract.proto
index 5e8abde3c..cdb93c74b 100644
--- a/proto/def/contract.proto
+++ b/proto/def/contract.proto
@@ -39,14 +39,8 @@ message EgressConfig {
   // backoffPolicy is the retry backoff policy (linear, exponential).
   BackoffPolicy backoffPolicy = 3;
 
-  // backoffDelay is the delay before retrying.
-  // More information on Duration format:
-  //  - https://www.iso.org/iso-8601-date-and-time-format.html
-  //  - https://en.wikipedia.org/wiki/ISO_8601
-  //
-  // For linear policy, backoff delay is backoffDelay*<numberOfRetries>.
-  // For exponential policy, backoff delay is backoffDelay*2^<numberOfRetries>.
-  string backoffDelay = 4;
+  // backoffDelay is the delay before retrying in milliseconds.
+  uint64 backoffDelay = 4;
 }
 
 message Egress {
diff --git a/vendor/github.com/rickb777/date/period/arithmetic.go b/vendor/github.com/rickb777/date/period/arithmetic.go
new file mode 100644
index 000000000..6141c0917
--- /dev/null
+++ b/vendor/github.com/rickb777/date/period/arithmetic.go
@@ -0,0 +1,106 @@
+// Copyright 2015 Rick Beton. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package period
+
+import (
+	"time"
+)
+
+// Add adds two periods together. Use this method along with Negate in order to subtract periods.
+//
+// The result is not normalised and may overflow arithmetically (to make this unlikely, use Normalise on
+// the inputs before adding them).
+func (period Period) Add(that Period) Period {
+	return Period{
+		period.years + that.years,
+		period.months + that.months,
+		period.days + that.days,
+		period.hours + that.hours,
+		period.minutes + that.minutes,
+		period.seconds + that.seconds,
+	}
+}
+
+//-------------------------------------------------------------------------------------------------
+
+// AddTo adds the period to a time, returning the result.
+// A flag is also returned that is true when the conversion was precise and false otherwise.
+//
+// When the period specifies hours, minutes and seconds only, the result is precise.
+// Also, when the period specifies whole years, months and days (i.e. without fractions), the
+// result is precise. However, when years, months or days contains fractions, the result
+// is only an approximation (it assumes that all days are 24 hours and every year is 365.2425
+// days, as per Gregorian calendar rules).
+func (period Period) AddTo(t time.Time) (time.Time, bool) {
+	wholeYears := (period.years % 10) == 0
+	wholeMonths := (period.months % 10) == 0
+	wholeDays := (period.days % 10) == 0
+
+	if wholeYears && wholeMonths && wholeDays {
+		// in this case, time.AddDate provides an exact solution
+		stE3 := totalSecondsE3(period)
+		t1 := t.AddDate(int(period.years/10), int(period.months/10), int(period.days/10))
+		return t1.Add(stE3 * time.Millisecond), true
+	}
+
+	d, precise := period.Duration()
+	return t.Add(d), precise
+}
+
+//-------------------------------------------------------------------------------------------------
+
+// Scale a period by a multiplication factor. Obviously, this can both enlarge and shrink it,
+// and change the sign if negative. The result is normalised, but integer overflows are silently
+// ignored.
+//
+// Bear in mind that the internal representation is limited by fixed-point arithmetic with two
+// decimal places; each field is only int16.
+//
+// Known issue: scaling by a large reduction factor (i.e. much less than one) doesn't work properly.
+func (period Period) Scale(factor float32) Period {
+	result, _ := period.ScaleWithOverflowCheck(factor)
+	return result
+}
+
+// ScaleWithOverflowCheck a period by a multiplication factor. Obviously, this can both enlarge and shrink it,
+// and change the sign if negative. The result is normalised. An error is returned if integer overflow
+// happened.
+//
+// Bear in mind that the internal representation is limited by fixed-point arithmetic with one
+// decimal place; each field is only int16.
+//
+// Known issue: scaling by a large reduction factor (i.e. much less than one) doesn't work properly.
+func (period Period) ScaleWithOverflowCheck(factor float32) (Period, error) {
+	ap, neg := period.absNeg()
+
+	if -0.5 < factor && factor < 0.5 {
+		d, pr1 := ap.Duration()
+		mul := float64(d) * float64(factor)
+		p2, pr2 := NewOf(time.Duration(mul))
+		return p2.Normalise(pr1 && pr2), nil
+	}
+
+	y := int(float32(ap.years) * factor)
+	m := int(float32(ap.months) * factor)
+	d := int(float32(ap.days) * factor)
+	hh := int(float32(ap.hours) * factor)
+	mm := int(float32(ap.minutes) * factor)
+	ss := int(float32(ap.seconds) * factor)
+
+	p64 := &period64{years: y, months: m, days: d, hours: hh, minutes: mm, seconds: ss, neg: neg}
+	return p64.normalise64(true).toPeriod()
+}
+
+// RationalScale scales a period by a rational multiplication factor. Obviously, this can both enlarge and shrink it,
+// and change the sign if negative. The result is normalised. An error is returned if integer overflow
+// happened.
+//
+// If the divisor is zero, a panic will arise.
+//
+// Bear in mind that the internal representation is limited by fixed-point arithmetic with two
+// decimal places; each field is only int16.
+//func (period Period) RationalScale(multiplier, divisor int) (Period, error) {
+//	return period.rationalScale64(int64(multiplier), int64(divisor))
+//}
diff --git a/vendor/github.com/rickb777/date/period/designator.go b/vendor/github.com/rickb777/date/period/designator.go
new file mode 100644
index 000000000..6d836ec4c
--- /dev/null
+++ b/vendor/github.com/rickb777/date/period/designator.go
@@ -0,0 +1,55 @@
+// Copyright 2015 Rick Beton. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package period
+
+type ymdDesignator byte
+type hmsDesignator byte
+
+const (
+	Year  ymdDesignator = 'Y'
+	Month ymdDesignator = 'M'
+	Week  ymdDesignator = 'W'
+	Day   ymdDesignator = 'D'
+
+	Hour   hmsDesignator = 'H'
+	Minute hmsDesignator = 'M'
+	Second hmsDesignator = 'S'
+)
+
+func (d ymdDesignator) IsOneOf(xx ...ymdDesignator) bool {
+	for _, x := range xx {
+		if x == d {
+			return true
+		}
+	}
+	return false
+}
+
+func (d ymdDesignator) IsNotOneOf(xx ...ymdDesignator) bool {
+	for _, x := range xx {
+		if x == d {
+			return false
+		}
+	}
+	return true
+}
+
+func (d hmsDesignator) IsOneOf(xx ...hmsDesignator) bool {
+	for _, x := range xx {
+		if x == d {
+			return true
+		}
+	}
+	return false
+}
+
+func (d hmsDesignator) IsNotOneOf(xx ...hmsDesignator) bool {
+	for _, x := range xx {
+		if x == d {
+			return false
+		}
+	}
+	return true
+}
diff --git a/vendor/github.com/rickb777/date/period/format.go b/vendor/github.com/rickb777/date/period/format.go
index 32cb6a831..df517b58a 100644
--- a/vendor/github.com/rickb777/date/period/format.go
+++ b/vendor/github.com/rickb777/date/period/format.go
@@ -5,25 +5,32 @@
 package period
 
 import (
-	"bytes"
 	"fmt"
+	"io"
 	"strings"
 
 	"github.com/rickb777/plural"
 )
 
 // Format converts the period to human-readable form using the default localisation.
+// Multiples of 7 days are shown as weeks.
 func (period Period) Format() string {
 	return period.FormatWithPeriodNames(PeriodYearNames, PeriodMonthNames, PeriodWeekNames, PeriodDayNames, PeriodHourNames, PeriodMinuteNames, PeriodSecondNames)
 }
 
+// FormatWithoutWeeks converts the period to human-readable form using the default localisation.
+// Multiples of 7 days are not shown as weeks.
+func (period Period) FormatWithoutWeeks() string {
+	return period.FormatWithPeriodNames(PeriodYearNames, PeriodMonthNames, plural.Plurals{}, PeriodDayNames, PeriodHourNames, PeriodMinuteNames, PeriodSecondNames)
+}
+
 // FormatWithPeriodNames converts the period to human-readable form in a localisable way.
 func (period Period) FormatWithPeriodNames(yearNames, monthNames, weekNames, dayNames, hourNames, minNames, secNames plural.Plurals) string {
 	period = period.Abs()
 
 	parts := make([]string, 0)
-	parts = appendNonBlank(parts, yearNames.FormatFloat(absFloat10(period.years)))
-	parts = appendNonBlank(parts, monthNames.FormatFloat(absFloat10(period.months)))
+	parts = appendNonBlank(parts, yearNames.FormatFloat(float10(period.years)))
+	parts = appendNonBlank(parts, monthNames.FormatFloat(float10(period.months)))
 
 	if period.days > 0 || (period.IsZero()) {
 		if len(weekNames) > 0 {
@@ -34,15 +41,15 @@ func (period Period) FormatWithPeriodNames(yearNames, monthNames, weekNames, day
 				parts = appendNonBlank(parts, weekNames.FormatInt(int(weeks)))
 			}
 			if mdays > 0 || weeks == 0 {
-				parts = appendNonBlank(parts, dayNames.FormatFloat(absFloat10(mdays)))
+				parts = appendNonBlank(parts, dayNames.FormatFloat(float10(mdays)))
 			}
 		} else {
-			parts = appendNonBlank(parts, dayNames.FormatFloat(absFloat10(period.days)))
+			parts = appendNonBlank(parts, dayNames.FormatFloat(float10(period.days)))
 		}
 	}
-	parts = appendNonBlank(parts, hourNames.FormatFloat(absFloat10(period.hours)))
-	parts = appendNonBlank(parts, minNames.FormatFloat(absFloat10(period.minutes)))
-	parts = appendNonBlank(parts, secNames.FormatFloat(absFloat10(period.seconds)))
+	parts = appendNonBlank(parts, hourNames.FormatFloat(float10(period.hours)))
+	parts = appendNonBlank(parts, minNames.FormatFloat(float10(period.minutes)))
+	parts = appendNonBlank(parts, secNames.FormatFloat(float10(period.seconds)))
 
 	return strings.Join(parts, ", ")
 }
@@ -79,50 +86,54 @@ var PeriodSecondNames = plural.FromZero("", "%v second", "%v seconds")
 
 // String converts the period to ISO-8601 form.
 func (period Period) String() string {
-	if period.IsZero() {
+	return period.toPeriod64("").String()
+}
+
+func (p64 period64) String() string {
+	if p64 == (period64{}) {
 		return "P0D"
 	}
 
-	buf := &bytes.Buffer{}
-	if period.Sign() < 0 {
+	buf := &strings.Builder{}
+	if p64.neg {
 		buf.WriteByte('-')
 	}
 
 	buf.WriteByte('P')
 
-	if period.years != 0 {
-		fmt.Fprintf(buf, "%gY", absFloat10(period.years))
-	}
-	if period.months != 0 {
-		fmt.Fprintf(buf, "%gM", absFloat10(period.months))
-	}
-	if period.days != 0 {
-		if period.days%70 == 0 {
-			fmt.Fprintf(buf, "%gW", absFloat10(period.days/7))
+	writeField64(buf, p64.years, byte(Year))
+	writeField64(buf, p64.months, byte(Month))
+
+	if p64.days != 0 {
+		if p64.days%70 == 0 {
+			writeField64(buf, p64.days/7, byte(Week))
 		} else {
-			fmt.Fprintf(buf, "%gD", absFloat10(period.days))
+			writeField64(buf, p64.days, byte(Day))
 		}
 	}
-	if period.hours != 0 || period.minutes != 0 || period.seconds != 0 {
+
+	if p64.hours != 0 || p64.minutes != 0 || p64.seconds != 0 {
 		buf.WriteByte('T')
 	}
-	if period.hours != 0 {
-		fmt.Fprintf(buf, "%gH", absFloat10(period.hours))
-	}
-	if period.minutes != 0 {
-		fmt.Fprintf(buf, "%gM", absFloat10(period.minutes))
-	}
-	if period.seconds != 0 {
-		fmt.Fprintf(buf, "%gS", absFloat10(period.seconds))
-	}
+
+	writeField64(buf, p64.hours, byte(Hour))
+	writeField64(buf, p64.minutes, byte(Minute))
+	writeField64(buf, p64.seconds, byte(Second))
 
 	return buf.String()
 }
 
-func absFloat10(v int16) float32 {
-	f := float32(v) / 10
-	if v < 0 {
-		return -f
+func writeField64(w io.Writer, field int, designator byte) {
+	if field != 0 {
+		if field%10 != 0 {
+			fmt.Fprintf(w, "%g", float32(field)/10)
+		} else {
+			fmt.Fprintf(w, "%d", field/10)
+		}
+		w.(io.ByteWriter).WriteByte(designator)
 	}
-	return f
+}
+
+func float10(v int16) float32 {
+	return float32(v) / 10
 }
diff --git a/vendor/github.com/rickb777/date/period/parse.go b/vendor/github.com/rickb777/date/period/parse.go
index 52b91f142..c6fb8d0df 100644
--- a/vendor/github.com/rickb777/date/period/parse.go
+++ b/vendor/github.com/rickb777/date/period/parse.go
@@ -12,8 +12,10 @@ import (
 
 // MustParse is as per Parse except that it panics if the string cannot be parsed.
 // This is intended for setup code; don't use it for user inputs.
-func MustParse(value string) Period {
-	d, err := Parse(value)
+// By default, the value is normalised.
+// Normalisation can be disabled using the optional flag.
+func MustParse(value string, normalise ...bool) Period {
+	d, err := Parse(value, normalise...)
 	if err != nil {
 		panic(err)
 	}
@@ -24,34 +26,27 @@ func MustParse(value string) Period {
 //
 // In addition, a plus or minus sign can precede the period, e.g. "-P10D"
 //
-// The value is normalised, e.g. multiple of 12 months become years so "P24M"
-// is the same as "P2Y". However, this is done without loss of precision, so
-// for example whole numbers of days do not contribute to the months tally
+// By default, the value is normalised, e.g. multiple of 12 months become years
+// so "P24M" is the same as "P2Y". However, this is done without loss of precision,
+// so for example whole numbers of days do not contribute to the months tally
 // because the number of days per month is variable.
 //
+// Normalisation can be disabled using the optional flag.
+//
 // The zero value can be represented in several ways: all of the following
 // are equivalent: "P0Y", "P0M", "P0W", "P0D", "PT0H", PT0M", PT0S", and "P0".
 // The canonical zero is "P0D".
-func Parse(period string) (Period, error) {
-	return ParseWithNormalise(period, true)
+func Parse(period string, normalise ...bool) (Period, error) {
+	return ParseWithNormalise(period, len(normalise) == 0 || normalise[0])
 }
 
 // ParseWithNormalise parses strings that specify periods using ISO-8601 rules
 // with an option to specify whether to normalise parsed period components.
 //
-// In addition, a plus or minus sign can precede the period, e.g. "-P10D"
-
-// The returned value is only normalised when normalise is set to `true`, and
-// normalisation will convert e.g. multiple of 12 months into years so "P24M"
-// is the same as "P2Y". However, this is done without loss of precision, so
-// for example whole numbers of days do not contribute to the months tally
-// because the number of days per month is variable.
-//
-// The zero value can be represented in several ways: all of the following
-// are equivalent: "P0Y", "P0M", "P0W", "P0D", "PT0H", PT0M", PT0S", and "P0".
-// The canonical zero is "P0D".
+// This method is deprecated and should not be used. It may be removed in a
+// future version.
 func ParseWithNormalise(period string, normalise bool) (Period, error) {
-	if period == "" {
+	if period == "" || period == "-" || period == "+" {
 		return Period{}, fmt.Errorf("cannot parse a blank string as a period")
 	}
 
@@ -59,122 +54,184 @@ func ParseWithNormalise(period string, normalise bool) (Period, error) {
 		return Period{}, nil
 	}
 
-	result := period64{}
-	pcopy := period
-	if pcopy[0] == '-' {
-		result.neg = true
-		pcopy = pcopy[1:]
-	} else if pcopy[0] == '+' {
-		pcopy = pcopy[1:]
+	p64, err := parse(period, normalise)
+	if err != nil {
+		return Period{}, err
 	}
+	return p64.toPeriod()
+}
 
-	if pcopy[0] != 'P' {
-		return Period{}, fmt.Errorf("expected 'P' period mark at the start: %s", period)
+func parse(period string, normalise bool) (*period64, error) {
+	neg := false
+	remaining := period
+	if remaining[0] == '-' {
+		neg = true
+		remaining = remaining[1:]
+	} else if remaining[0] == '+' {
+		remaining = remaining[1:]
 	}
-	pcopy = pcopy[1:]
 
-	st := parseState{period, pcopy, false, nil}
-	t := strings.IndexByte(pcopy, 'T')
-	if t >= 0 {
-		st.pcopy = pcopy[t+1:]
+	if remaining[0] != 'P' {
+		return nil, fmt.Errorf("%s: expected 'P' period mark at the start", period)
+	}
+	remaining = remaining[1:]
 
-		result.hours, st = parseField(st, 'H')
-		if st.err != nil {
-			return Period{}, fmt.Errorf("expected a number before the 'H' marker: %s", period)
-		}
+	var number, weekValue, prevFraction int
+	result := &period64{input: period, neg: neg}
+	var years, months, weeks, days, hours, minutes, seconds itemState
+	var designator, prevDesignator byte
+	var err error
+	nComponents := 0
 
-		result.minutes, st = parseField(st, 'M')
-		if st.err != nil {
-			return Period{}, fmt.Errorf("expected a number before the 'M' marker: %s", period)
-		}
+	years, months, weeks, days = Armed, Armed, Armed, Armed
 
-		result.seconds, st = parseField(st, 'S')
-		if st.err != nil {
-			return Period{}, fmt.Errorf("expected a number before the 'S' marker: %s", period)
-		}
+	isHMS := false
+	for len(remaining) > 0 {
+		if remaining[0] == 'T' {
+			if isHMS {
+				return nil, fmt.Errorf("%s: 'T' designator cannot occur more than once", period)
+			}
+			isHMS = true
 
-		if len(st.pcopy) != 0 {
-			return Period{}, fmt.Errorf("unexpected remaining components %s: %s", st.pcopy, period)
-		}
+			years, months, weeks, days = Unready, Unready, Unready, Unready
+			hours, minutes, seconds = Armed, Armed, Armed
 
-		st.pcopy = pcopy[:t]
-	}
+			remaining = remaining[1:]
 
-	result.years, st = parseField(st, 'Y')
-	if st.err != nil {
-		return Period{}, fmt.Errorf("expected a number before the 'Y' marker: %s", period)
-	}
-	result.months, st = parseField(st, 'M')
-	if st.err != nil {
-		return Period{}, fmt.Errorf("expected a number before the 'M' marker: %s", period)
-	}
-	weeks, st := parseField(st, 'W')
-	if st.err != nil {
-		return Period{}, fmt.Errorf("expected a number before the 'W' marker: %s", period)
+		} else {
+			number, designator, remaining, err = parseNextField(remaining, period)
+			if err != nil {
+				return nil, err
+			}
+
+			fraction := number % 10
+			if prevFraction != 0 && fraction != 0 {
+				return nil, fmt.Errorf("%s: '%c' & '%c' only the last field can have a fraction", period, prevDesignator, designator)
+			}
+
+			switch designator {
+			case 'Y':
+				years, err = years.testAndSet(number, 'Y', result, &result.years)
+			case 'W':
+				weeks, err = weeks.testAndSet(number, 'W', result, &weekValue)
+			case 'D':
+				days, err = days.testAndSet(number, 'D', result, &result.days)
+			case 'H':
+				hours, err = hours.testAndSet(number, 'H', result, &result.hours)
+			case 'S':
+				seconds, err = seconds.testAndSet(number, 'S', result, &result.seconds)
+			case 'M':
+				if isHMS {
+					minutes, err = minutes.testAndSet(number, 'M', result, &result.minutes)
+				} else {
+					months, err = months.testAndSet(number, 'M', result, &result.months)
+				}
+			default:
+				return nil, fmt.Errorf("%s: expected a number not '%c'", period, designator)
+			}
+			nComponents++
+
+			if err != nil {
+				return nil, err
+			}
+
+			prevFraction = fraction
+			prevDesignator = designator
+		}
 	}
 
-	days, st := parseField(st, 'D')
-	if st.err != nil {
-		return Period{}, fmt.Errorf("expected a number before the 'D' marker: %s", period)
+	if nComponents == 0 {
+		return nil, fmt.Errorf("%s: expected 'Y', 'M', 'W', 'D', 'H', 'M', or 'S' designator", period)
 	}
 
-	if len(st.pcopy) != 0 {
-		return Period{}, fmt.Errorf("unexpected remaining components %s: %s", st.pcopy, period)
+	result.days += weekValue * 7
+
+	if normalise {
+		result = result.normalise64(true)
 	}
 
-	result.days = weeks*7 + days
-	//fmt.Printf("%#v\n", st)
+	return result, nil
+}
 
-	if !st.ok {
-		return Period{}, fmt.Errorf("expected 'Y', 'M', 'W', 'D', 'H', 'M', or 'S' marker: %s", period)
-	}
+//-------------------------------------------------------------------------------------------------
 
-	if normalise {
-		return result.normalise64(true).toPeriod(), nil
+type itemState int
+
+const (
+	Unready itemState = iota
+	Armed
+	Set
+)
+
+func (i itemState) testAndSet(number int, designator byte, result *period64, value *int) (itemState, error) {
+	switch i {
+	case Unready:
+		return i, fmt.Errorf("%s: '%c' designator cannot occur here", result.input, designator)
+	case Set:
+		return i, fmt.Errorf("%s: '%c' designator cannot occur more than once", result.input, designator)
 	}
 
-	return result.toPeriod(), nil
+	*value = number
+	return Set, nil
 }
 
-type parseState struct {
-	period, pcopy string
-	ok            bool
-	err           error
-}
+//-------------------------------------------------------------------------------------------------
 
-func parseField(st parseState, mark byte) (int64, parseState) {
-	//fmt.Printf("%c %#v\n", mark, st)
-	r := int64(0)
-	m := strings.IndexByte(st.pcopy, mark)
-	if m > 0 {
-		r, st.err = parseDecimalFixedPoint(st.pcopy[:m], st.period)
-		if st.err != nil {
-			return 0, st
-		}
-		st.pcopy = st.pcopy[m+1:]
-		st.ok = true
+func parseNextField(str, original string) (int, byte, string, error) {
+	i := scanDigits(str)
+	if i < 0 {
+		return 0, 0, "", fmt.Errorf("%s: missing designator at the end", original)
 	}
-	return r, st
+
+	des := str[i]
+	number, err := parseDecimalNumber(str[:i], original, des)
+	return number, des, str[i+1:], err
 }
 
-// Fixed-point three decimal places
-func parseDecimalFixedPoint(s, original string) (int64, error) {
-	//was := s
-	dec := strings.IndexByte(s, '.')
+// Fixed-point one decimal place
+func parseDecimalNumber(number, original string, des byte) (int, error) {
+	dec := strings.IndexByte(number, '.')
 	if dec < 0 {
-		dec = strings.IndexByte(s, ',')
+		dec = strings.IndexByte(number, ',')
 	}
 
+	var integer, fraction int
+	var err error
 	if dec >= 0 {
-		dp := len(s) - dec
-		if dp > 1 {
-			s = s[:dec] + s[dec+1:dec+2]
-		} else {
-			s = s[:dec] + s[dec+1:] + "0"
+		integer, err = strconv.Atoi(number[:dec])
+		if err == nil {
+			number = number[dec+1:]
+			switch len(number) {
+			case 0: // skip
+			case 1:
+				fraction, err = strconv.Atoi(number)
+			default:
+				fraction, err = strconv.Atoi(number[:1])
+			}
 		}
 	} else {
-		s = s + "0"
+		integer, err = strconv.Atoi(number)
+	}
+
+	if err != nil {
+		return 0, fmt.Errorf("%s: expected a number but found '%c'", original, des)
+	}
+
+	return integer*10 + fraction, err
+}
+
+// scanDigits finds the first non-digit byte after a given starting point.
+// Note that it does not care about runes or UTF-8 encoding; it assumes that
+// a period string is always valid ASCII as well as UTF-8.
+func scanDigits(s string) int {
+	for i, c := range s {
+		if !isDigit(c) {
+			return i
+		}
 	}
+	return -1
+}
 
-	return strconv.ParseInt(s, 10, 64)
+func isDigit(c rune) bool {
+	return ('0' <= c && c <= '9') || c == '.' || c == ','
 }
diff --git a/vendor/github.com/rickb777/date/period/period.go b/vendor/github.com/rickb777/date/period/period.go
index a9b39c6a0..d097e9c80 100644
--- a/vendor/github.com/rickb777/date/period/period.go
+++ b/vendor/github.com/rickb777/date/period/period.go
@@ -9,14 +9,14 @@ import (
 	"time"
 )
 
-const daysPerYearE4 int64 = 3652425   // 365.2425 days by the Gregorian rule
-const daysPerMonthE4 int64 = 304369   // 30.4369 days per month
-const daysPerMonthE6 int64 = 30436875 // 30.436875 days per month
+const daysPerYearE4 = 3652425   // 365.2425 days by the Gregorian rule
+const daysPerMonthE4 = 304369   // 30.4369 days per month
+const daysPerMonthE6 = 30436875 // 30.436875 days per month
 
-const oneE4 int64 = 10000
-const oneE5 int64 = 100000
-const oneE6 int64 = 1000000
-const oneE7 int64 = 10000000
+const oneE4 = 10000
+const oneE5 = 100000
+const oneE6 = 1000000
+const oneE7 = 10000000
 
 const hundredMs = 100 * time.Millisecond
 
@@ -52,6 +52,8 @@ type Period struct {
 // need to.
 //
 // All the parameters must have the same sign (otherwise a panic occurs).
+// Because this implementation uses int16 internally, the paramters must
+// be within the range ± 2^16 / 10.
 func NewYMD(years, months, days int) Period {
 	return New(years, months, days, 0, 0, 0)
 }
@@ -61,6 +63,8 @@ func NewYMD(years, months, days int) Period {
 // if you need to.
 //
 // All the parameters must have the same sign (otherwise a panic occurs).
+// Because this implementation uses int16 internally, the paramters must
+// be within the range ± 2^16 / 10.
 func NewHMS(hours, minutes, seconds int) Period {
 	return New(0, 0, 0, hours, minutes, seconds)
 }
@@ -82,8 +86,6 @@ func New(years, months, days, hours, minutes, seconds int) Period {
 		years, months, days, hours, minutes, seconds))
 }
 
-// TODO NewFloat
-
 // NewOf converts a time duration to a Period, and also indicates whether the conversion is precise.
 // Any time duration that spans more than ± 3276 hours will be approximated by assuming that there
 // are 24 hours per day, 365.2425 days per year (as per Gregorian calendar rules), and a month
@@ -142,15 +144,15 @@ func NewOf(duration time.Duration) (p Period, precise bool) {
 // computations applied to the period can only be precise if they concern either the date (year, month,
 // day) part, or the clock (hour, minute, second) part, but not both.
 func Between(t1, t2 time.Time) (p Period) {
-	if t1.Location() != t2.Location() {
-		t2 = t2.In(t1.Location())
-	}
-
 	sign := 1
 	if t2.Before(t1) {
 		t1, t2, sign = t2, t1, -1
 	}
 
+	if t1.Location() != t2.Location() {
+		t2 = t2.In(t1.Location())
+	}
+
 	year, month, day, hour, min, sec, hundredth := daysDiff(t1, t2)
 
 	if sign < 0 {
@@ -171,9 +173,9 @@ func daysDiff(t1, t2 time.Time) (year, month, day, hour, min, sec, hundredth int
 
 	day = int(duration / (24 * time.Hour))
 
-	hour = int(hh2 - hh1)
-	min = int(mm2 - mm1)
-	sec = int(ss2 - ss1)
+	hour = hh2 - hh1
+	min = mm2 - mm1
+	sec = ss2 - ss1
 	hundredth = (t2.Nanosecond() - t1.Nanosecond()) / 100000000
 
 	// Normalize negative values
@@ -248,15 +250,22 @@ func (period Period) OnlyHMS() Period {
 
 // Abs converts a negative period to a positive one.
 func (period Period) Abs() Period {
-	return Period{absInt16(period.years), absInt16(period.months), absInt16(period.days),
-		absInt16(period.hours), absInt16(period.minutes), absInt16(period.seconds)}
+	a, _ := period.absNeg()
+	return a
 }
 
-func absInt16(v int16) int16 {
-	if v < 0 {
-		return -v
+func (period Period) absNeg() (Period, bool) {
+	if period.IsNegative() {
+		return period.Negate(), true
 	}
-	return v
+	return period, false
+}
+
+func (period Period) condNegate(neg bool) Period {
+	if neg {
+		return period.Negate()
+	}
+	return period
 }
 
 // Negate changes the sign of the period.
@@ -264,45 +273,11 @@ func (period Period) Negate() Period {
 	return Period{-period.years, -period.months, -period.days, -period.hours, -period.minutes, -period.seconds}
 }
 
-// Add adds two periods together. Use this method along with Negate in order to subtract periods.
-//
-// The result is not normalised and may overflow arithmetically (to make this unlikely, use Normalise on
-// the inputs before adding them).
-func (period Period) Add(that Period) Period {
-	return Period{
-		period.years + that.years,
-		period.months + that.months,
-		period.days + that.days,
-		period.hours + that.hours,
-		period.minutes + that.minutes,
-		period.seconds + that.seconds,
-	}
-}
-
-// Scale a period by a multiplication factor. Obviously, this can both enlarge and shrink it,
-// and change the sign if negative. The result is normalised.
-//
-// Bear in mind that the internal representation is limited by fixed-point arithmetic with one
-// decimal place; each field is only int16.
-//
-// Known issue: scaling by a large reduction factor (i.e. much less than one) doesn't work properly.
-func (period Period) Scale(factor float32) Period {
-
-	if -0.5 < factor && factor < 0.5 {
-		d, pr1 := period.Duration()
-		mul := float64(d) * float64(factor)
-		p2, pr2 := NewOf(time.Duration(mul))
-		return p2.Normalise(pr1 && pr2)
+func absInt16(v int16) int16 {
+	if v < 0 {
+		return -v
 	}
-
-	y := int64(float32(period.years) * factor)
-	m := int64(float32(period.months) * factor)
-	d := int64(float32(period.days) * factor)
-	hh := int64(float32(period.hours) * factor)
-	mm := int64(float32(period.minutes) * factor)
-	ss := int64(float32(period.seconds) * factor)
-
-	return (&period64{y, m, d, hh, mm, ss, false}).normalise64(true).toPeriod()
+	return v
 }
 
 // Years gets the whole number of years in the period.
@@ -422,29 +397,7 @@ func (period Period) SecondsFloat() float32 {
 	return float32(period.seconds) / 10
 }
 
-// AddTo adds the period to a time, returning the result.
-// A flag is also returned that is true when the conversion was precise and false otherwise.
-//
-// When the period specifies hours, minutes and seconds only, the result is precise.
-// Also, when the period specifies whole years, months and days (i.e. without fractions), the
-// result is precise. However, when years, months or days contains fractions, the result
-// is only an approximation (it assumes that all days are 24 hours and every year is 365.2425
-// days, as per Gregorian calendar rules).
-func (period Period) AddTo(t time.Time) (time.Time, bool) {
-	wholeYears := (period.years % 10) == 0
-	wholeMonths := (period.months % 10) == 0
-	wholeDays := (period.days % 10) == 0
-
-	if wholeYears && wholeMonths && wholeDays {
-		// in this case, time.AddDate provides an exact solution
-		stE3 := totalSecondsE3(period)
-		t1 := t.AddDate(int(period.years/10), int(period.months/10), int(period.days/10))
-		return t1.Add(stE3 * time.Millisecond), true
-	}
-
-	d, precise := period.Duration()
-	return t.Add(d), precise
-}
+//-------------------------------------------------------------------------------------------------
 
 // DurationApprox converts a period to the equivalent duration in nanoseconds.
 // When the period specifies hours, minutes and seconds only, the result is precise.
@@ -514,7 +467,8 @@ func (period Period) TotalMonthsApprox() int {
 //
 // Because the number of hours per day is imprecise (due to daylight savings etc), and because
 // the number of days per month is variable in the Gregorian calendar, there is a reluctance
-// to transfer time too or from the days element. To give control over this, there are two modes.
+// to transfer time to or from the days element, or to transfer days to or from the months
+// element. To give control over this, there are two modes.
 //
 // In precise mode:
 // Multiples of 60 seconds become minutes.
@@ -527,220 +481,139 @@ func (period Period) TotalMonthsApprox() int {
 //
 // Note that leap seconds are disregarded: every minute is assumed to have 60 seconds.
 func (period Period) Normalise(precise bool) Period {
-	const limit = 32670 - (32670 / 60)
-
-	// can we use a quicker algorithm for HHMMSS with int16 arithmetic?
-	if period.years == 0 && period.months == 0 &&
-		(!precise || period.days == 0) &&
-		period.hours > -limit && period.hours < limit {
-
-		return period.normaliseHHMMSS(precise)
-	}
-
-	// can we use a quicker algorithm for YYMM with int16 arithmetic?
-	if (period.years != 0 || period.months != 0) && //period.months%10 == 0 &&
-		period.days == 0 && period.hours == 0 && period.minutes == 0 && period.seconds == 0 {
-
-		return period.normaliseYYMM()
-	}
-
-	// do things the no-nonsense way using int64 arithmetic
-	return period.toPeriod64().normalise64(precise).toPeriod()
+	n, _ := period.toPeriod64("").normalise64(precise).toPeriod()
+	return n
 }
 
-func (period Period) normaliseHHMMSS(precise bool) Period {
-	s := period.Sign()
-	ap := period.Abs()
-
-	// remember that the fields are all fixed-point 1E1
-	ap.minutes += (ap.seconds / 600) * 10
-	ap.seconds = ap.seconds % 600
-
-	ap.hours += (ap.minutes / 600) * 10
-	ap.minutes = ap.minutes % 600
-
-	// up to 36 hours stays as hours
-	if !precise && ap.hours > 360 {
-		ap.days += (ap.hours / 240) * 10
-		ap.hours = ap.hours % 240
-	}
-
-	d10 := ap.days % 10
-	if d10 != 0 && (ap.hours != 0 || ap.minutes != 0 || ap.seconds != 0) {
-		ap.hours += d10 * 24
-		ap.days -= d10
-	}
-
-	hh10 := ap.hours % 10
-	if hh10 != 0 {
-		ap.minutes += hh10 * 60
-		ap.hours -= hh10
-	}
-
-	mm10 := ap.minutes % 10
-	if mm10 != 0 {
-		ap.seconds += mm10 * 60
-		ap.minutes -= mm10
-	}
-
-	if s < 0 {
-		return ap.Negate()
+// Simplify applies some heuristic simplifications with the objective of reducing the number
+// of non-zero fields and thus making the rendered form simpler. It should be applied to
+// a normalised period, otherwise the results may be unpredictable.
+//
+// Note that months and days are never combined, due to the variability of month lengths.
+// Days and hours are only combined when imprecise behaviour is selected; this is due to
+// daylight savings transitions, during which there are more than or fewer than 24 hours
+// per day.
+//
+// The following transformation rules are applied in order:
+//
+// * P1YnM becomes 12+n months for 0 < n <= 6
+// * P1DTnH becomes 24+n hours for 0 < n <= 6 (unless precise is true)
+// * PT1HnM becomes 60+n minutes for 0 < n <= 10
+// * PT1MnS becomes 60+n seconds for 0 < n <= 10
+//
+// At each step, if a fraction exists and would affect the calculation, the transformations
+// stop. Also, when not precise,
+//
+// * for periods of at least ten years, month proper fractions are discarded
+// * for periods of at least a year, day proper fractions are discarded
+// * for periods of at least a month, hour proper fractions are discarded
+// * for periods of at least a day, minute proper fractions are discarded
+// * for periods of at least an hour, second proper fractions are discarded
+//
+// The thresholds can be set using the varargs th parameter. By default, the thresholds a,
+// b, c, d are 6 months, 6 hours, 10 minutes, 10 seconds respectively as listed in the rules
+// above.
+//
+// * No thresholds is equivalent to 6, 6, 10, 10.
+// * A single threshold a is equivalent to a, a, a, a.
+// * Two thresholds a, b are equivalent to a, a, b, b.
+// * Three thresholds a, b, c are equivalent to a, b, c, c.
+// * Four thresholds a, b, c, d are used as provided.
+//
+func (period Period) Simplify(precise bool, th ...int) Period {
+	switch len(th) {
+	case 0:
+		return period.doSimplify(precise, 60, 60, 100, 100)
+	case 1:
+		return period.doSimplify(precise, int16(th[0]*10), int16(th[0]*10), int16(th[0]*10), int16(th[0]*10))
+	case 2:
+		return period.doSimplify(precise, int16(th[0]*10), int16(th[0]*10), int16(th[1]*10), int16(th[1]*10))
+	case 3:
+		return period.doSimplify(precise, int16(th[0]*10), int16(th[1]*10), int16(th[2]*10), int16(th[2]*10))
+	default:
+		return period.doSimplify(precise, int16(th[0]*10), int16(th[1]*10), int16(th[2]*10), int16(th[3]*10))
 	}
-	return ap
 }
 
-func (period Period) normaliseYYMM() Period {
-	s := period.Sign()
-	ap := period.Abs()
-
-	// remember that the fields are all fixed-point 1E1
-	if ap.months > 129 {
-		ap.years += (ap.months / 120) * 10
-		ap.months = ap.months % 120
+func (period Period) doSimplify(precise bool, a, b, c, d int16) Period {
+	if period.years%10 != 0 {
+		return period
 	}
 
-	y10 := ap.years % 10
-	if y10 != 0 && (ap.years < 10 || ap.months != 0) {
-		ap.months += y10 * 12
-		ap.years -= y10
-	}
+	ap, neg := period.absNeg()
 
-	if s < 0 {
-		return ap.Negate()
+	// single year is dropped if there are some months
+	if ap.years == 10 &&
+		0 < ap.months && ap.months <= a &&
+		ap.days == 0 {
+		ap.months += 120
+		ap.years = 0
 	}
-	return ap
-}
 
-//-------------------------------------------------------------------------------------------------
-
-// used for stages in arithmetic
-type period64 struct {
-	years, months, days, hours, minutes, seconds int64
-	neg                                          bool
-}
-
-func (period Period) toPeriod64() *period64 {
-	return &period64{
-		int64(period.years), int64(period.months), int64(period.days),
-		int64(period.hours), int64(period.minutes), int64(period.seconds),
-		false,
-	}
-}
-
-func (p *period64) toPeriod() Period {
-	if p.neg {
-		return Period{
-			int16(-p.years), int16(-p.months), int16(-p.days),
-			int16(-p.hours), int16(-p.minutes), int16(-p.seconds),
+	if ap.months%10 != 0 {
+		// month fraction is dropped for periods of at least ten years (1:120)
+		months := ap.months / 10
+		if !precise && ap.years >= 100 && months == 0 {
+			ap.months = 0
 		}
+		return ap.condNegate(neg)
 	}
 
-	return Period{
-		int16(p.years), int16(p.months), int16(p.days),
-		int16(p.hours), int16(p.minutes), int16(p.seconds),
-	}
-}
-
-func (p *period64) normalise64(precise bool) *period64 {
-	return p.abs().rippleUp(precise).moveFractionToRight()
-}
-
-func (p *period64) abs() *period64 {
-
-	if !p.neg {
-		if p.years < 0 {
-			p.years = -p.years
-			p.neg = true
-		}
-
-		if p.months < 0 {
-			p.months = -p.months
-			p.neg = true
-		}
-
-		if p.days < 0 {
-			p.days = -p.days
-			p.neg = true
-		}
-
-		if p.hours < 0 {
-			p.hours = -p.hours
-			p.neg = true
+	if ap.days%10 != 0 {
+		// day fraction is dropped for periods of at least a year (1:365)
+		days := ap.days / 10
+		if !precise && (ap.years > 0 || ap.months >= 120) && days == 0 {
+			ap.days = 0
 		}
-
-		if p.minutes < 0 {
-			p.minutes = -p.minutes
-			p.neg = true
-		}
-
-		if p.seconds < 0 {
-			p.seconds = -p.seconds
-			p.neg = true
-		}
-	}
-	return p
-}
-
-func (p *period64) rippleUp(precise bool) *period64 {
-	// remember that the fields are all fixed-point 1E1
-
-	p.minutes = p.minutes + (p.seconds/600)*10
-	p.seconds = p.seconds % 600
-
-	p.hours = p.hours + (p.minutes/600)*10
-	p.minutes = p.minutes % 600
-
-	// 32670-(32670/60)-(32670/3600) = 32760 - 546 - 9.1 = 32204.9
-	if !precise || p.hours > 32204 {
-		p.days += (p.hours / 240) * 10
-		p.hours = p.hours % 240
+		return ap.condNegate(neg)
 	}
 
-	if !precise || p.days > 32760 {
-		dE6 := p.days * oneE6
-		p.months += dE6 / daysPerMonthE6
-		p.days = (dE6 % daysPerMonthE6) / oneE6
+	if !precise && ap.days == 10 &&
+		ap.years == 0 &&
+		ap.months == 0 &&
+		0 < ap.hours && ap.hours <= b {
+		ap.hours += 240
+		ap.days = 0
 	}
 
-	p.years = p.years + (p.months/120)*10
-	p.months = p.months % 120
-
-	return p
-}
-
-// moveFractionToRight applies the rule that only the smallest field is permitted to have a decimal fraction.
-func (p *period64) moveFractionToRight() *period64 {
-	// remember that the fields are all fixed-point 1E1
-
-	y10 := p.years % 10
-	if y10 != 0 && (p.months != 0 || p.days != 0 || p.hours != 0 || p.minutes != 0 || p.seconds != 0) {
-		p.months += y10 * 12
-		p.years = (p.years / 10) * 10
+	if ap.hours%10 != 0 {
+		// hour fraction is dropped for periods of at least a month (1:720)
+		hours := ap.hours / 10
+		if !precise && (ap.years > 0 || ap.months > 0 || ap.days >= 300) && hours == 0 {
+			ap.hours = 0
+		}
+		return ap.condNegate(neg)
 	}
 
-	m10 := p.months % 10
-	if m10 != 0 && (p.days != 0 || p.hours != 0 || p.minutes != 0 || p.seconds != 0) {
-		p.days += (m10 * daysPerMonthE6) / oneE6
-		p.months = (p.months / 10) * 10
+	if ap.hours == 10 &&
+		0 < ap.minutes && ap.minutes <= c {
+		ap.minutes += 600
+		ap.hours = 0
 	}
 
-	d10 := p.days % 10
-	if d10 != 0 && (p.hours != 0 || p.minutes != 0 || p.seconds != 0) {
-		p.hours += d10 * 24
-		p.days = (p.days / 10) * 10
+	if ap.minutes%10 != 0 {
+		// minute fraction is dropped for periods of at least a day (1:1440)
+		minutes := ap.minutes / 10
+		if !precise && (ap.years > 0 || ap.months > 0 || ap.days > 0 || ap.hours >= 240) && minutes == 0 {
+			ap.minutes = 0
+		}
+		return ap.condNegate(neg)
 	}
 
-	hh10 := p.hours % 10
-	if hh10 != 0 && (p.minutes != 0 || p.seconds != 0) {
-		p.minutes += hh10 * 60
-		p.hours = (p.hours / 10) * 10
+	if ap.minutes == 10 &&
+		ap.hours == 0 &&
+		0 < ap.seconds && ap.seconds <= d {
+		ap.seconds += 600
+		ap.minutes = 0
 	}
 
-	mm10 := p.minutes % 10
-	if mm10 != 0 && p.seconds != 0 {
-		p.seconds += mm10 * 60
-		p.minutes = (p.minutes / 10) * 10
+	if ap.seconds%10 != 0 {
+		// second fraction is dropped for periods of at least an hour (1:3600)
+		seconds := ap.seconds / 10
+		if !precise && (ap.years > 0 || ap.months > 0 || ap.days > 0 || ap.hours > 0 || ap.minutes >= 600) && seconds == 0 {
+			ap.seconds = 0
+		}
 	}
 
-	return p
+	return ap.condNegate(neg)
 }
diff --git a/vendor/github.com/rickb777/date/period/period64.go b/vendor/github.com/rickb777/date/period/period64.go
new file mode 100644
index 000000000..f3ea7892b
--- /dev/null
+++ b/vendor/github.com/rickb777/date/period/period64.go
@@ -0,0 +1,142 @@
+package period
+
+import (
+	"fmt"
+	"math"
+	"strings"
+)
+
+// used for stages in arithmetic
+type period64 struct {
+	// always positive values
+	years, months, days, hours, minutes, seconds int
+	// true if the period is negative
+	neg   bool
+	input string
+}
+
+func (period Period) toPeriod64(input string) *period64 {
+	if period.IsNegative() {
+		return &period64{
+			years: int(-period.years), months: int(-period.months), days: int(-period.days),
+			hours: int(-period.hours), minutes: int(-period.minutes), seconds: int(-period.seconds),
+			neg:   true,
+			input: input,
+		}
+	}
+	return &period64{
+		years: int(period.years), months: int(period.months), days: int(period.days),
+		hours: int(period.hours), minutes: int(period.minutes), seconds: int(period.seconds),
+		input: input,
+	}
+}
+
+func (p64 *period64) toPeriod() (Period, error) {
+	var f []string
+	if p64.years > math.MaxInt16 {
+		f = append(f, "years")
+	}
+	if p64.months > math.MaxInt16 {
+		f = append(f, "months")
+	}
+	if p64.days > math.MaxInt16 {
+		f = append(f, "days")
+	}
+	if p64.hours > math.MaxInt16 {
+		f = append(f, "hours")
+	}
+	if p64.minutes > math.MaxInt16 {
+		f = append(f, "minutes")
+	}
+	if p64.seconds > math.MaxInt16 {
+		f = append(f, "seconds")
+	}
+
+	if len(f) > 0 {
+		if p64.input == "" {
+			p64.input = p64.String()
+		}
+		return Period{}, fmt.Errorf("%s: integer overflow occurred in %s", p64.input, strings.Join(f, ","))
+	}
+
+	if p64.neg {
+		return Period{
+			int16(-p64.years), int16(-p64.months), int16(-p64.days),
+			int16(-p64.hours), int16(-p64.minutes), int16(-p64.seconds),
+		}, nil
+	}
+
+	return Period{
+		int16(p64.years), int16(p64.months), int16(p64.days),
+		int16(p64.hours), int16(p64.minutes), int16(p64.seconds),
+	}, nil
+}
+
+func (p64 *period64) normalise64(precise bool) *period64 {
+	return p64.rippleUp(precise).moveFractionToRight()
+}
+
+func (p64 *period64) rippleUp(precise bool) *period64 {
+	// remember that the fields are all fixed-point 1E1
+
+	p64.minutes += (p64.seconds / 600) * 10
+	p64.seconds = p64.seconds % 600
+
+	p64.hours += (p64.minutes / 600) * 10
+	p64.minutes = p64.minutes % 600
+
+	// 32670-(32670/60)-(32670/3600) = 32760 - 546 - 9.1 = 32204.9
+	if !precise || p64.hours > 32204 {
+		p64.days += (p64.hours / 240) * 10
+		p64.hours = p64.hours % 240
+	}
+
+	if !precise || p64.days > 32760 {
+		dE6 := p64.days * oneE5
+		p64.months += (dE6 / daysPerMonthE6) * 10
+		p64.days = (dE6 % daysPerMonthE6) / oneE5
+	}
+
+	p64.years += (p64.months / 120) * 10
+	p64.months = p64.months % 120
+
+	return p64
+}
+
+// moveFractionToRight attempts to remove fractions in higher-order fields by moving their value to the
+// next-lower-order field. For example, fractional years become months.
+func (p64 *period64) moveFractionToRight() *period64 {
+	// remember that the fields are all fixed-point 1E1
+
+	y10 := p64.years % 10
+	if y10 != 0 && (p64.months != 0 || p64.days != 0 || p64.hours != 0 || p64.minutes != 0 || p64.seconds != 0) {
+		p64.months += y10 * 12
+		p64.years = (p64.years / 10) * 10
+	}
+
+	m10 := p64.months % 10
+	if m10 != 0 && (p64.days != 0 || p64.hours != 0 || p64.minutes != 0 || p64.seconds != 0) {
+		p64.days += (m10 * daysPerMonthE6) / oneE6
+		p64.months = (p64.months / 10) * 10
+	}
+
+	d10 := p64.days % 10
+	if d10 != 0 && (p64.hours != 0 || p64.minutes != 0 || p64.seconds != 0) {
+		p64.hours += d10 * 24
+		p64.days = (p64.days / 10) * 10
+	}
+
+	hh10 := p64.hours % 10
+	if hh10 != 0 && (p64.minutes != 0 || p64.seconds != 0) {
+		p64.minutes += hh10 * 60
+		p64.hours = (p64.hours / 10) * 10
+	}
+
+	mm10 := p64.minutes % 10
+	if mm10 != 0 && p64.seconds != 0 {
+		p64.seconds += mm10 * 60
+		p64.minutes = (p64.minutes / 10) * 10
+	}
+
+	return p64
+}
diff --git a/vendor/github.com/rickb777/plural/go.mod b/vendor/github.com/rickb777/plural/go.mod
index 3f1ef90dd..fe9f8b863 100644
--- a/vendor/github.com/rickb777/plural/go.mod
+++ b/vendor/github.com/rickb777/plural/go.mod
@@ -1,3 +1,3 @@
 module github.com/rickb777/plural
 
-go 1.14
+go 1.15
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 76cc6bc8f..c0eaaba55 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -289,10 +289,10 @@ github.com/prometheus/statsd_exporter/pkg/mapper
 github.com/prometheus/statsd_exporter/pkg/mapper/fsm
 # github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0
 github.com/rcrowley/go-metrics
-# github.com/rickb777/date v1.13.0
+# github.com/rickb777/date v1.14.1
 ## explicit
 github.com/rickb777/date/period
-# github.com/rickb777/plural v1.2.1
+# github.com/rickb777/plural v1.2.2
 github.com/rickb777/plural
 # github.com/robfig/cron/v3 v3.0.1
 github.com/robfig/cron/v3
@@ -378,7 +378,6 @@ golang.org/x/lint/golint
 golang.org/x/mod/module
 golang.org/x/mod/semver
 # golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0
-## explicit
 golang.org/x/net/context
 golang.org/x/net/context/ctxhttp
 golang.org/x/net/http/httpguts


Test patch:
diff --git a/control-plane/pkg/config/env_config_test.go b/control-plane/pkg/config/env_config_test.go
index 7df164dce..54328c467 100644
--- a/control-plane/pkg/config/env_config_test.go
+++ b/control-plane/pkg/config/env_config_test.go
@@ -1,28 +1,41 @@
+/*
+ * Copyright 2020 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package config
 
 import (
+	"io"
 	"os"
 	"reflect"
 	"testing"
 )
 
 func TestGetEnvConfig(t *testing.T) {
-	type args struct {
-		prefix string
-	}
 
 	tests := []struct {
-		name    string
-		args    args
-		setEnv  func()
-		want    *Env
-		wantErr bool
+		name        string
+		prefix      string
+		validations []ValidationOption
+		setEnv      func()
+		want        *Env
+		wantErr     bool
 	}{
 		{
-			name: "broker prefix",
-			args: args{
-				prefix: "BROKER",
-			},
+			name:   "broker prefix",
+			prefix: "BROKER",
 			want: &Env{
 				DataPlaneConfigMapNamespace: "knative-eventing",
 				DataPlaneConfigMapName:      "kafka-brokers-triggers",
@@ -42,10 +55,8 @@ func TestGetEnvConfig(t *testing.T) {
 			wantErr: false,
 		},
 		{
-			name: "sink prefix",
-			args: args{
-				prefix: "SINK",
-			},
+			name:   "sink prefix",
+			prefix: "SINK",
 			want: &Env{
 				DataPlaneConfigMapNamespace: "knative-eventing",
 				DataPlaneConfigMapName:      "kafka-sinks",
@@ -65,10 +76,8 @@ func TestGetEnvConfig(t *testing.T) {
 			wantErr: false,
 		},
 		{
-			name: "missing required variable - SINK_DATA_PLANE_CONFIG_MAP_NAMESPACE",
-			args: args{
-				prefix: "SINK",
-			},
+			name:   "missing required variable - SINK_DATA_PLANE_CONFIG_MAP_NAMESPACE",
+			prefix: "SINK",
 			setEnv: func() {
 				_ = os.Setenv("SINK_DATA_PLANE_CONFIG_MAP_NAME", "kafka-sinks")
 				_ = os.Setenv("SINK_GENERAL_CONFIG_MAP_NAME", "kafka-config")
@@ -79,9 +88,37 @@ func TestGetEnvConfig(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "invalid backoff delay",
-			args: args{
-				prefix: "BROKER",
+			name:   "invalid backoff delay 0",
+			prefix: "BROKER",
+			validations: []ValidationOption{
+				func(env Env) error {
+					if env.DefaultBackoffDelayMs == 0 {
+						return io.EOF
+					}
+					return nil
+				},
+			},
+			setEnv: func() {
+				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_MAP_NAMESPACE", "knative-eventing")
+				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_MAP_NAME", "kafka-brokers-triggers")
+				_ = os.Setenv("BROKER_GENERAL_CONFIG_MAP_NAME", "kafka-config")
+				_ = os.Setenv("BROKER_INGRESS_NAME", "kafka-broker-ingress")
+				_ = os.Setenv("BROKER_SYSTEM_NAMESPACE", "knative-eventing")
+				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_FORMAT", "json")
+				_ = os.Setenv("BROKER_DEFAULT_BACKOFF_DELAY_MS", "0")
+			},
+			wantErr: true,
+		},
+		{
+			name:   "invalid backoff delay unset",
+			prefix: "BROKER",
+			validations: []ValidationOption{
+				func(env Env) error {
+					if env.DefaultBackoffDelayMs == 0 {
+						return io.EOF
+					}
+					return nil
+				},
 			},
 			setEnv: func() {
 				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_MAP_NAMESPACE", "knative-eventing")
@@ -90,7 +127,6 @@ func TestGetEnvConfig(t *testing.T) {
 				_ = os.Setenv("BROKER_INGRESS_NAME", "kafka-broker-ingress")
 				_ = os.Setenv("BROKER_SYSTEM_NAMESPACE", "knative-eventing")
 				_ = os.Setenv("BROKER_DATA_PLANE_CONFIG_FORMAT", "json")
-				_ = os.Setenv("BROKER_DEFAULT_BACKOFF_DELAY", "PTT")
 			},
 			wantErr: true,
 		},
@@ -101,7 +137,7 @@ func TestGetEnvConfig(t *testing.T) {
 			tt.setEnv()
 			defer os.Clearenv()
 
-			got, err := GetEnvConfig(tt.args.prefix)
+			got, err := GetEnvConfig(tt.prefix, tt.validations...)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("GetEnvConfig() error = %v, wantErr %v", err, tt.wantErr)
 				return
diff --git a/control-plane/pkg/core/config/utils_test.go b/control-plane/pkg/core/config/utils_test.go
index 0e90edeeb..11c2b57a3 100644
--- a/control-plane/pkg/core/config/utils_test.go
+++ b/control-plane/pkg/core/config/utils_test.go
@@ -104,25 +104,66 @@ func TestBackoffDelayFromString(t *testing.T) {
 	tests := []struct {
 		name         string
 		backoffDelay *string
-		defaultDelay string
-		want         string
+		defaultDelay uint64
+		want         uint64
+		wantError    bool
 	}{
 		{
 			name:         "happy case",
 			backoffDelay: pointer.StringPtr("PT2S"),
-			defaultDelay: "PT1S",
-			want:         "PT2S",
+			defaultDelay: 1000,
+			want:         2000,
+		},
+		{
+			name:         "happy case fractional",
+			backoffDelay: pointer.StringPtr("PT0.2S"),
+			defaultDelay: 1000,
+			want:         200,
+		},
+		{
+			name:         "happy case fractional - 2 decimals",
+			backoffDelay: pointer.StringPtr("PT0.25S"),
+			defaultDelay: 1000,
+			want:         200, // The library we use don't support more than one decimal, it should be 250 though.
+		},
+		{
+			name:         "happy case fractional minutes",
+			backoffDelay: pointer.StringPtr("PT1M0.2S"),
+			defaultDelay: 1000,
+			want:         1*60*1000 + 200,
+		},
+		{
+			name:         "milliseconds",
+			backoffDelay: pointer.StringPtr("PT0.001S"),
+			defaultDelay: 1000,
+			want:         0,
+		},
+		{
+			name:         "microseconds",
+			backoffDelay: pointer.StringPtr("PT0.000001S"),
+			defaultDelay: 1000,
+			want:         0,
+		},
+		{
+			name:         "65 seconds",
+			backoffDelay: pointer.StringPtr("PT65S"),
+			defaultDelay: 1000,
+			want:         1 * 65 * 1000,
 		},
 		{
 			name:         "default",
-			defaultDelay: "PT1S",
-			want:         "PT1S",
+			defaultDelay: 1000,
+			want:         1000,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := BackoffDelayFromString(tt.backoffDelay, tt.defaultDelay); got != tt.want {
-				t.Errorf("BackoffDelayFromString() = %v, want %v", got, tt.want)
+			got, err := BackoffDelayFromISO8601String(tt.backoffDelay, tt.defaultDelay)
+			if (err != nil) != tt.wantError {
+				t.Errorf("wantError = %v got %v", tt.wantError, err)
+			}
+			if got != tt.want {
+				t.Errorf("BackoffDelayFromISO8601String() = %v, want %v", got, tt.want)
 			}
 		})
 	}
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 4b59bd384..62cd01487 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -1140,7 +1140,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 								DeadLetter:    "http://test-service.test-service-namespace.svc.cluster.local/",
 								Retry:         10,
 								BackoffPolicy: contract.BackoffPolicy_Exponential,
-								BackoffDelay:  "PT2S",
+								BackoffDelay:  2000,
 							},
 						},
 					},
@@ -1204,7 +1204,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 								DeadLetter:    "http://test-service.test-service-namespace.svc.cluster.local/",
 								Retry:         10,
 								BackoffPolicy: contract.BackoffPolicy_Linear,
-								BackoffDelay:  "PT2S",
+								BackoffDelay:  2000,
 							},
 						},
 					},
@@ -1329,7 +1329,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 								DeadLetter:    "http://test-service.test-service-namespace.svc.cluster.local/",
 								Retry:         10,
 								BackoffPolicy: contract.BackoffPolicy_Linear,
-								BackoffDelay:  configs.DefaultBackoffDelay,
+								BackoffDelay:  configs.DefaultBackoffDelayMs,
 							},
 						},
 					},
diff --git a/control-plane/pkg/reconciler/broker/controller_test.go b/control-plane/pkg/reconciler/broker/controller_test.go
index bdbe312e0..71ba497a0 100644
--- a/control-plane/pkg/reconciler/broker/controller_test.go
+++ b/control-plane/pkg/reconciler/broker/controller_test.go
@@ -73,3 +73,34 @@ func TestNewController(t *testing.T) {
 		t.Error("failed to create controller: <nil>")
 	}
 }
+
+func TestValidateDefaultBackoffDelayMs(t *testing.T) {
+
+	tests := []struct {
+		name    string
+		env     config.Env
+		wantErr bool
+	}{
+		{
+			name: "0 want error",
+			env: config.Env{
+				DefaultBackoffDelayMs: 0,
+			},
+			wantErr: true,
+		},
+		{
+			name: "non 0 no error",
+			env: config.Env{
+				DefaultBackoffDelayMs: 1,
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := ValidateDefaultBackoffDelayMs(tt.env); (err != nil) != tt.wantErr {
+				t.Errorf("ValidateDefaultBackoffDelayMs() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/testing/factory.go b/control-plane/pkg/reconciler/testing/factory.go
index 296c028d3..e8602ffc5 100644
--- a/control-plane/pkg/reconciler/testing/factory.go
+++ b/control-plane/pkg/reconciler/testing/factory.go
@@ -55,7 +55,7 @@ var DefaultConfigs = &broker.Configs{
 		IngressName:                 "kafka-broker-receiver",
 		SystemNamespace:             "knative-eventing",
 		DataPlaneConfigFormat:       base.Json,
-		DefaultBackoffDelay:         "PT1S",
+		DefaultBackoffDelayMs:       1000,
 	},
 
 	BootstrapServers: "",
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactoryTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactoryTest.java
index ed7b4faba..0252fe441 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactoryTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactoryTest.java
@@ -27,6 +27,7 @@ import static org.mockito.Mockito.mock;
 
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract.BackoffPolicy;
+import dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EgressConfig;
 import dev.knative.eventing.kafka.broker.core.wrappers.Egress;
 import dev.knative.eventing.kafka.broker.core.wrappers.EventMatcher;
 import dev.knative.eventing.kafka.broker.core.wrappers.Filter;
@@ -96,7 +97,7 @@ public class HttpConsumerVerticleFactoryTest {
         @Override
         public DataPlaneContract.EgressConfig egressConfig() {
           return DataPlaneContract.EgressConfig.newBuilder()
-            .setBackoffDelay("PT1S")
+            .setBackoffDelay(1000)
             .setBackoffPolicy(BackoffPolicy.Exponential)
             .setRetry(10)
             .setDeadLetter("http://localhost:43257")
@@ -224,4 +225,56 @@ public class HttpConsumerVerticleFactoryTest {
         });
     });
   }
+
+  @Test
+  public void linearBackoffPolicy() {
+
+    final var policy = HttpConsumerVerticleFactory.computeRetryPolicy(EgressConfig.newBuilder()
+      .setRetry(10)
+      .setBackoffPolicy(BackoffPolicy.Linear)
+      .setBackoffDelay(100)
+      .build());
+
+    final var delay = policy.apply(5);
+
+    assertThat(delay).isEqualTo(100 * 5);
+  }
+
+  @Test
+  public void exponentialBackoffPolicy() {
+
+    final var policy = HttpConsumerVerticleFactory.computeRetryPolicy(EgressConfig.newBuilder()
+      .setRetry(10)
+      .setBackoffPolicy(BackoffPolicy.Exponential)
+      .setBackoffDelay(100)
+      .build());
+
+    final var delay = policy.apply(5);
+
+    assertThat(delay).isEqualTo((long) (100 * Math.pow(2, 5)));
+  }
+
+  @Test
+  public void exponentialBackoffPolicyByDefault() {
+
+    final var policy = HttpConsumerVerticleFactory.computeRetryPolicy(EgressConfig.newBuilder()
+      .setRetry(10)
+      .setBackoffPolicy(BackoffPolicy.Exponential)
+      .setBackoffDelay(100)
+      .build());
+
+    final var delay = policy.apply(5);
+
+    assertThat(delay).isEqualTo((long) (100 * Math.pow(2, 5)));
+  }
+
+  @Test
+  public void noRetry() {
+
+    final var policy = HttpConsumerVerticleFactory.computeRetryPolicy(null);
+
+    final var delay = policy.apply(Double.valueOf(Math.random()).intValue());
+
+    assertThat(delay).isEqualTo(0);
+  }
 }
diff --git a/test/data-plane/library.sh b/test/data-plane/library.sh
index aaf390e44..e16168d4a 100755
--- a/test/data-plane/library.sh
+++ b/test/data-plane/library.sh
@@ -41,11 +41,10 @@ readonly receiver="${KNATIVE_KAFKA_BROKER_RECEIVER:-knative-kafka-broker-receive
 readonly dispatcher="${KNATIVE_KAFKA_BROKER_DISPATCHER:-knative-kafka-broker-dispatcher}"
 readonly sink="${KNATIVE_KAFKA_SINK_RECEIVER:-knative-kafka-sink-receiver}"
 
-# The BASE_IMAGE for the running image should be based on the same base of JAVA_IMAGE
-# because jlink generates a jdk linked to the libc available on the local machine
-# adoptopenjdk uses ubuntu:bionic as base
-readonly JAVA_IMAGE=docker.io/adoptopenjdk:14-jdk-hotspot
-readonly BASE_IMAGE=docker.io/ubuntu:bionic
+# The BASE_IMAGE must have system libraries (libc, zlib, etc) compatible with the JAVA_IMAGE because
+# Jlink generates a jdk linked to the same system libraries available on the base images.
+readonly BASE_IMAGE=gcr.io/distroless/java-debian10:base-nonroot    # Based on debian:buster
+readonly JAVA_IMAGE=docker.io/adoptopenjdk/openjdk14:debian         # Based on debian:buster
 
 readonly RECEIVER_JAR="receiver-1.0-SNAPSHOT.jar"
 readonly RECEIVER_DIRECTORY=receiver
diff --git a/test/test_images/kafka-consumer/consumer.go b/test/test_images/kafka-consumer/consumer.go
index 086d91d0a..2eaa8c57a 100644
--- a/test/test_images/kafka-consumer/consumer.go
+++ b/test/test_images/kafka-consumer/consumer.go
@@ -18,6 +18,7 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"log"
 	"strings"
@@ -26,9 +27,9 @@ import (
 	protocolkafka "github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2"
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
-	cetest "github.com/cloudevents/sdk-go/v2/test"
 	"github.com/google/uuid"
 	"github.com/kelseyhightower/envconfig"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"knative.dev/pkg/signals"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
@@ -47,7 +48,8 @@ func main() {
 		log.Fatal("Failed to process environment variables", err)
 	}
 
-	log.Println(envConfig)
+	j, _ := json.MarshalIndent(envConfig, "", " ")
+	log.Println(string(j))
 
 	consumerConfig := sarama.NewConfig()
 	consumerConfig.Version = sarama.V2_0_0_0
@@ -94,22 +96,21 @@ func main() {
 	}()
 
 	ids := strings.Split(envConfig.IDS, ",")
-	matchers := make([]cetest.EventMatcher, 0, len(ids))
-	for _, id := range ids {
-		matchers = append(matchers, cetest.HasId(id))
-	}
-
-	matcher := cetest.AnyOf(matchers...)
+	set := sets.NewString(ids...)
+	defer func() {
+		log.Println("Remaining", set)
+	}()
 
-	for i := 0; i < len(ids); i++ {
+	for set.Len() == 0 {
 		select {
 		case err := <-exitError:
 			log.Fatal("Failed to consume", err)
 		case e := <-events:
-			if err := matcher(e); err != nil {
-				log.Fatalf("Failed to match event: %s\n", e.String())
+			if set.Has(e.ID()) {
+				set.Delete(e.ID())
+				log.Printf("Event matching received: %s\nRemaining: %v\n", e.ID(), set)
 			} else {
-				log.Println("event matching received", e.String())
+				log.Fatalf("Failed to match event: %s\n%v\n", e.String(), set.List())
 			}
 		}
 	}
@@ -120,11 +121,11 @@ type handler struct {
 	encoding binding.Encoding
 }
 
-func (h *handler) Setup(session sarama.ConsumerGroupSession) error {
+func (h *handler) Setup(_ sarama.ConsumerGroupSession) error {
 	return nil
 }
 
-func (h *handler) Cleanup(session sarama.ConsumerGroupSession) error {
+func (h *handler) Cleanup(_ sarama.ConsumerGroupSession) error {
 	return nil
 }
 
@@ -148,8 +149,6 @@ func (h *handler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama
 			return fmt.Errorf("failed to convert message to event: %w", err)
 		}
 
-		log.Println(event.String())
-
 		h.events <- *event
 	}