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

Issue:
Use Jlink to reduce container image size
**Problem**
A short explanation of the problem, including relevant restrictions.

**[Persona:](https://github.com/knative/eventing/blob/master/docs/personas.md)** 
Which persona is this feature for?
- System Integrator

**Exit Criteria**
Reduce container image size ( < 300MB )

**Time Estimate (optional):** 
How many developer-days do you think this may take to resolve?
1-4

**Additional context (optional)**
- https://docs.oracle.com/en/java/javase/11/tools/jlink.html


Pull request:
Jlink to reduce container image size
Fixes #10 

Now image size (receiver) is 184Mb, compared to previous 478Mb

This is a first pass on the issue, there are some mods picked by jdeps that we could eventually remove (like java.sql, that we don't use but we bring in b/c of jackson-databind). Also, we're on ubuntu base, we could drill down the image even further porting to alpine (but that requires musl libc and we need to figure out how to make that working with jlink).

## Proposed Changes

- Use jlink to create a smaller jdk, in order to reduce image size of our deliverables

Patch:
diff --git a/control-plane/config/500-controller.yaml b/control-plane/config/500-controller.yaml
index 3eedb95ad..62a1040ad 100644
--- a/control-plane/config/500-controller.yaml
+++ b/control-plane/config/500-controller.yaml
@@ -86,6 +86,8 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
+            - name: BROKER_DEFAULT_BACKOFF_DELAY
+              value: "PT1S"
             - name: SYSTEM_NAMESPACE
               valueFrom:
                 fieldRef:
diff --git a/control-plane/pkg/config/env_config.go b/control-plane/pkg/config/env_config.go
index 19d8cb0b8..95c25c283 100644
--- a/control-plane/pkg/config/env_config.go
+++ b/control-plane/pkg/config/env_config.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	"github.com/kelseyhightower/envconfig"
+	"github.com/rickb777/date/period"
 )
 
 type Env struct {
@@ -29,6 +30,7 @@ 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"`
 }
 
 func GetEnvConfig(prefix string) (*Env, error) {
@@ -38,9 +40,20 @@ func GetEnvConfig(prefix string) (*Env, error) {
 		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)
+		}
+	}
+
 	return env, nil
 }
 
 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/core/config/utils.go b/control-plane/pkg/core/config/utils.go
index 18bb8eec8..25e9c65bf 100644
--- a/control-plane/pkg/core/config/utils.go
+++ b/control-plane/pkg/core/config/utils.go
@@ -19,6 +19,8 @@ package config
 import (
 	"fmt"
 
+	duck "knative.dev/eventing/pkg/apis/duck/v1"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
@@ -39,3 +41,33 @@ func ContentModeFromString(mode string) contract.ContentMode {
 		))
 	}
 }
+
+// BackoffPolicyFromString returns the BackoffPolicy from the given string.
+//
+// Default value is contract.BackoffPolicy_Exponential.
+func BackoffPolicyFromString(backoffPolicy *duck.BackoffPolicyType) contract.BackoffPolicy {
+	if backoffPolicy == nil {
+		return contract.BackoffPolicy_Exponential
+	}
+
+	bp := *backoffPolicy
+	switch bp {
+	case duck.BackoffPolicyLinear:
+		return contract.BackoffPolicy_Linear
+	case duck.BackoffPolicyExponential: // The linter complains for missing case in switch
+		return contract.BackoffPolicy_Exponential
+	default:
+		return contract.BackoffPolicy_Exponential
+	}
+}
+
+// BackoffDelayFromString returns the BackoffDelay from the given string.
+//
+// Default value is the specified defaultDelay.
+func BackoffDelayFromString(backoffDelay *string, defaultDelay string) string {
+	if backoffDelay == nil {
+		return defaultDelay
+	}
+
+	return *backoffDelay
+}
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index ea04057ae..28765aca0 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -317,17 +317,37 @@ func (r *Reconciler) getBrokerResource(ctx context.Context, topic string, broker
 		BootstrapServers: config.GetBootstrapServers(),
 	}
 
-	if broker.Spec.Delivery != nil && broker.Spec.Delivery.DeadLetterSink != nil {
-		deadLetterSinkURL, err := r.Resolver.URIFromDestinationV1(ctx, *broker.Spec.Delivery.DeadLetterSink, broker)
-		if err != nil {
-			return nil, fmt.Errorf("failed to resolve broker.Spec.Deliver.DeadLetterSink: %w", err)
+	delivery := broker.Spec.Delivery
+	if delivery != nil {
+
+		if delivery.DeadLetterSink != nil {
+
+			deadLetterSinkURL, err := r.Resolver.URIFromDestinationV1(ctx, *delivery.DeadLetterSink, broker)
+			if err != nil {
+				return nil, fmt.Errorf("failed to resolve broker.Spec.Deliver.DeadLetterSink: %w", err)
+			}
+
+			ensureEgressConfig(res)
+			res.EgressConfig.DeadLetter = deadLetterSinkURL.String()
+		}
+
+		if delivery.Retry != nil {
+			ensureEgressConfig(res)
+			res.EgressConfig.Retry = uint32(*delivery.Retry)
+			res.EgressConfig.BackoffDelay = coreconfig.BackoffDelayFromString(delivery.BackoffDelay, "PT1S")
+			res.EgressConfig.BackoffPolicy = coreconfig.BackoffPolicyFromString(delivery.BackoffPolicy)
 		}
-		res.EgressConfig = &contract.EgressConfig{DeadLetter: deadLetterSinkURL.String()}
 	}
 
 	return res, nil
 }
 
+func ensureEgressConfig(res *contract.Resource) {
+	if res.EgressConfig == nil {
+		res.EgressConfig = &contract.EgressConfig{}
+	}
+}
+
 func (r *Reconciler) ConfigMapUpdated(ctx context.Context) func(configMap *corev1.ConfigMap) {
 
 	logger := logging.FromContext(ctx).Desugar()
diff --git a/data-plane/config/template/500-dispatcher.yaml b/data-plane/config/template/500-dispatcher.yaml
index 636724afe..cbbb6eba1 100644
--- a/data-plane/config/template/500-dispatcher.yaml
+++ b/data-plane/config/template/500-dispatcher.yaml
@@ -87,6 +87,26 @@ spec:
             - "-Dlogback.configurationFile=/etc/logging/config.xml"
             - "-jar"
             - "/app/app.jar"
+          livenessProbe:
+            failureThreshold: 3
+            httpGet:
+              port: 9090
+              path: /metrics
+              scheme: HTTP
+            initialDelaySeconds: 5
+            periodSeconds: 3
+            successThreshold: 1
+            timeoutSeconds: 1
+          readinessProbe:
+            failureThreshold: 3
+            httpGet:
+              port: 9090
+              path: /metrics
+              scheme: HTTP
+            initialDelaySeconds: 5
+            periodSeconds: 3
+            successThreshold: 1
+            timeoutSeconds: 1
           terminationMessagePolicy: FallbackToLogsOnError
           terminationMessagePath: /dev/temination-log
           securityContext:
@@ -101,7 +121,7 @@ spec:
           configMap:
             name: kafka-broker-brokers-triggers
         - name: cache
-          emptyDir: {}
+          emptyDir: { }
         - name: kafka-broker-config-logging
           configMap:
             name: kafka-config-logging
diff --git a/data-plane/docker/Dockerfile b/data-plane/docker/Dockerfile
index 4a7ebec3f..80291cad1 100644
--- a/data-plane/docker/Dockerfile
+++ b/data-plane/docker/Dockerfile
@@ -13,13 +13,18 @@
 # limitations under the License.
 
 ARG JAVA_IMAGE
-ARG APP_DIR
-ARG APP_JAR
+ARG BASE_IMAGE
 
 FROM ${JAVA_IMAGE} as builder
 
+ARG APP_DIR
+ARG APP_JAR
+
 WORKDIR /app
 
+# https://github.com/AdoptOpenJDK/openjdk-docker/issues/260
+RUN apt-get update && apt-get install -y binutils
+
 COPY pom.xml .
 COPY .editorconfig .
 COPY core/pom.xml core/pom.xml
@@ -29,24 +34,32 @@ COPY contract/pom.xml contract/pom.xml
 COPY tests/pom.xml tests/pom.xml
 COPY ./mvnw .
 COPY ./.mvn/wrapper .mvn/wrapper
+COPY docker/generate_jdk.sh .
 
-RUN ./mvnw install -DskipTests -Deditorconfig.skip --no-transfer-progress
+RUN ./mvnw install -pl=${APP_DIR} -am -DskipTests -Deditorconfig.skip --no-transfer-progress
 
 COPY . .
 
-RUN ./mvnw package -DskipTests -Deditorconfig.skip --no-transfer-progress
+RUN ./mvnw package -pl=${APP_DIR} -am -DskipTests -Deditorconfig.skip --no-transfer-progress
 
-FROM ${JAVA_IMAGE} as running
+RUN ./generate_jdk.sh /app/${APP_DIR}/target/${APP_JAR}
+
+FROM ${BASE_IMAGE} as running
 
 ARG APP_DIR
 ARG APP_JAR
-ARG JAVA_IMAGE
 
-RUN groupadd -g 999 appuser && useradd -r -u 999 -g appuser appuser
-RUN mkdir /tmp/vertx-cache && \
-      chown -R appuser:appuser /tmp/vertx-cache && \
-      mkdir /app && \
-      chown -R appuser:appuser /app
+# 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
+
+# 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
+
+# Set appuser and configure PATH
 USER appuser
 WORKDIR /app
-COPY --from=builder /app/${APP_DIR}/target/${APP_JAR} /app/app.jar
+ENV PATH="/app/jdk/bin:${PATH}"
diff --git a/data-plane/docker/generate_jdk.sh b/data-plane/docker/generate_jdk.sh
new file mode 100755
index 000000000..1f1f5f234
--- /dev/null
+++ b/data-plane/docker/generate_jdk.sh
@@ -0,0 +1,18 @@
+#!/usr/bin/env sh
+
+if [ $# -eq 0 ]
+  then
+    echo "No arguments supplied, You must provide the jar"
+fi
+
+echo "Computing mods for $1"
+MODS=$(jdeps -q --print-module-deps --ignore-missing-deps "$1" | sed -e 's/^[ \t]*//')
+
+echo "Computed mods = '$MODS'"
+
+# Patch adding the dns
+MODS="$MODS,jdk.naming.dns"
+
+echo "Patched modules shipped with the generated jdk = '$MODS'"
+jlink --verbose --no-header-files --no-man-pages --compress=2 --strip-debug --add-modules "$MODS" --output /app/jdk
+
diff --git a/go.mod b/go.mod
index c91148e69..16bb11795 100644
--- a/go.mod
+++ b/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/google/go-cmp v0.5.2
 	github.com/google/uuid v1.1.1
 	github.com/kelseyhightower/envconfig v1.4.0
+	github.com/rickb777/date v1.13.0
 	github.com/stretchr/testify v1.6.0
 	go.uber.org/zap v1.15.0
 	google.golang.org/protobuf v1.25.0
diff --git a/vendor/modules.txt b/vendor/modules.txt
index f5dde7b6d..cd396cee5 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -281,6 +281,7 @@ 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
+## explicit
 github.com/rickb777/date/period
 # github.com/rickb777/plural v1.2.1
 github.com/rickb777/plural


Test patch:
diff --git a/control-plane/pkg/config/env_config_test.go b/control-plane/pkg/config/env_config_test.go
index 83923bcf1..7df164dce 100644
--- a/control-plane/pkg/config/env_config_test.go
+++ b/control-plane/pkg/config/env_config_test.go
@@ -78,6 +78,22 @@ func TestGetEnvConfig(t *testing.T) {
 			},
 			wantErr: true,
 		},
+		{
+			name: "invalid backoff delay",
+			args: args{
+				prefix: "BROKER",
+			},
+			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", "PTT")
+			},
+			wantErr: true,
+		},
 	}
 
 	for _, tt := range tests {
diff --git a/control-plane/pkg/core/config/utils_test.go b/control-plane/pkg/core/config/utils_test.go
index 74c280150..0e90edeeb 100644
--- a/control-plane/pkg/core/config/utils_test.go
+++ b/control-plane/pkg/core/config/utils_test.go
@@ -19,6 +19,9 @@ package config
 import (
 	"testing"
 
+	"k8s.io/utils/pointer"
+	duck "knative.dev/eventing/pkg/apis/duck/v1"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
@@ -56,3 +59,71 @@ func TestContentModeFromString(t *testing.T) {
 		})
 	}
 }
+
+func TestBackoffPolicyFromString(t *testing.T) {
+	linerar := duck.BackoffPolicyLinear
+	exponential := duck.BackoffPolicyExponential
+	wrong := duck.BackoffPolicyType("default")
+	tests := []struct {
+		name          string
+		backoffPolicy *duck.BackoffPolicyType
+		want          contract.BackoffPolicy
+	}{
+		{
+			name:          "nil",
+			backoffPolicy: nil,
+			want:          contract.BackoffPolicy_Exponential,
+		},
+		{
+			name:          "exponential",
+			backoffPolicy: &exponential,
+			want:          contract.BackoffPolicy_Exponential,
+		},
+		{
+			name:          "linear",
+			backoffPolicy: &linerar,
+			want:          contract.BackoffPolicy_Linear,
+		},
+		{
+			name:          "default",
+			backoffPolicy: &wrong,
+			want:          contract.BackoffPolicy_Exponential,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := BackoffPolicyFromString(tt.backoffPolicy); got != tt.want {
+				t.Errorf("BackoffPolicyFromString() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestBackoffDelayFromString(t *testing.T) {
+
+	tests := []struct {
+		name         string
+		backoffDelay *string
+		defaultDelay string
+		want         string
+	}{
+		{
+			name:         "happy case",
+			backoffDelay: pointer.StringPtr("PT2S"),
+			defaultDelay: "PT1S",
+			want:         "PT2S",
+		},
+		{
+			name:         "default",
+			defaultDelay: "PT1S",
+			want:         "PT1S",
+		},
+	}
+	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)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index d8c32f55d..4b59bd384 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -24,6 +24,8 @@ import (
 	"testing"
 	"time"
 
+	"k8s.io/utils/pointer"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	"github.com/Shopify/sarama"
@@ -76,6 +78,9 @@ var (
 
 	createTopicError = fmt.Errorf("failed to create topic")
 	deleteTopicError = fmt.Errorf("failed to delete topic")
+
+	linear      = eventingduck.BackoffPolicyLinear
+	exponential = eventingduck.BackoffPolicyExponential
 )
 
 func TestBrokerReconciler(t *testing.T) {
@@ -1105,6 +1110,259 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal - with retry config - exponential",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+					WithRetry(pointer.Int32Ptr(10), &exponential, pointer.StringPtr("PT2S")),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Generation: 1,
+				}, &configs),
+				NewService(),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:               BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+							EgressConfig: &contract.EgressConfig{
+								DeadLetter:    "http://test-service.test-service-namespace.svc.cluster.local/",
+								Retry:         10,
+								BackoffPolicy: contract.BackoffPolicy_Exponential,
+								BackoffDelay:  "PT2S",
+							},
+						},
+					},
+					Generation: 2,
+				}),
+				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						WithRetry(pointer.Int32Ptr(10), &exponential, pointer.StringPtr("PT2S")),
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
+						BrokerTopicReady,
+						BrokerConfigParsed,
+						BrokerAddressable(&configs),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
+		{
+			Name: "Reconciled normal - with retry config - linear",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+					WithRetry(pointer.Int32Ptr(10), &linear, pointer.StringPtr("PT2S")),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Generation: 1,
+				}, &configs),
+				NewService(),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:               BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+							EgressConfig: &contract.EgressConfig{
+								DeadLetter:    "http://test-service.test-service-namespace.svc.cluster.local/",
+								Retry:         10,
+								BackoffPolicy: contract.BackoffPolicy_Linear,
+								BackoffDelay:  "PT2S",
+							},
+						},
+					},
+					Generation: 2,
+				}),
+				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						WithRetry(pointer.Int32Ptr(10), &linear, pointer.StringPtr("PT2S")),
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
+						BrokerTopicReady,
+						BrokerConfigParsed,
+						BrokerAddressable(&configs),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
+		{
+			Name: "Reconciled normal - with no retry num",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+					WithRetry(nil, &linear, pointer.StringPtr("PT2S")),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Generation: 1,
+				}, &configs),
+				NewService(),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:               BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+							EgressConfig: &contract.EgressConfig{
+								DeadLetter: "http://test-service.test-service-namespace.svc.cluster.local/",
+							},
+						},
+					},
+					Generation: 2,
+				}),
+				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						WithRetry(nil, &linear, pointer.StringPtr("PT2S")),
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
+						BrokerTopicReady,
+						BrokerConfigParsed,
+						BrokerAddressable(&configs),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
+		{
+			Name: "Reconciled normal - with retry config - no retry delay - use default delay",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+					WithRetry(pointer.Int32Ptr(10), &linear, nil),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Generation: 1,
+				}, &configs),
+				NewService(),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:               BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+							EgressConfig: &contract.EgressConfig{
+								DeadLetter:    "http://test-service.test-service-namespace.svc.cluster.local/",
+								Retry:         10,
+								BackoffPolicy: contract.BackoffPolicy_Linear,
+								BackoffDelay:  configs.DefaultBackoffDelay,
+							},
+						},
+					},
+					Generation: 2,
+				}),
+				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						WithRetry(pointer.Int32Ptr(10), &linear, nil),
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
+						BrokerTopicReady,
+						BrokerConfigParsed,
+						BrokerAddressable(&configs),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
 	}
 
 	for i := range table {
diff --git a/control-plane/pkg/reconciler/testing/factory.go b/control-plane/pkg/reconciler/testing/factory.go
index a7970ac99..296c028d3 100644
--- a/control-plane/pkg/reconciler/testing/factory.go
+++ b/control-plane/pkg/reconciler/testing/factory.go
@@ -50,11 +50,12 @@ const (
 var DefaultConfigs = &broker.Configs{
 
 	Env: config.Env{
-		DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
 		DataPlaneConfigMapNamespace: "knative-eventing",
+		DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
 		IngressName:                 "kafka-broker-receiver",
 		SystemNamespace:             "knative-eventing",
 		DataPlaneConfigFormat:       base.Json,
+		DefaultBackoffDelay:         "PT1S",
 	},
 
 	BootstrapServers: "",
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 913376d3a..8bf4a203e 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -80,19 +80,31 @@ func WithDelivery() func(*eventing.Broker) {
 	service := NewService()
 
 	return func(broker *eventing.Broker) {
-		broker.Spec.Delivery = &eventingduck.DeliverySpec{
-			DeadLetterSink: &duckv1.Destination{
-				Ref: &duckv1.KReference{
-					Kind:       service.Kind,
-					Namespace:  service.Namespace,
-					Name:       service.Name,
-					APIVersion: service.APIVersion,
-				},
+		if broker.Spec.Delivery == nil {
+			broker.Spec.Delivery = &eventingduck.DeliverySpec{}
+		}
+		broker.Spec.Delivery.DeadLetterSink = &duckv1.Destination{
+			Ref: &duckv1.KReference{
+				Kind:       service.Kind,
+				Namespace:  service.Namespace,
+				Name:       service.Name,
+				APIVersion: service.APIVersion,
 			},
 		}
 	}
 }
 
+func WithRetry(retry *int32, policy *eventingduck.BackoffPolicyType, delay *string) func(*eventing.Broker) {
+	return func(broker *eventing.Broker) {
+		if broker.Spec.Delivery == nil {
+			broker.Spec.Delivery = &eventingduck.DeliverySpec{}
+		}
+		broker.Spec.Delivery.Retry = retry
+		broker.Spec.Delivery.BackoffPolicy = policy
+		broker.Spec.Delivery.BackoffDelay = delay
+	}
+}
+
 func WithBrokerConfig(reference *duckv1.KReference) func(*eventing.Broker) {
 	return func(broker *eventing.Broker) {
 		broker.Spec.Config = reference
diff --git a/test/data-plane/library.sh b/test/data-plane/library.sh
index 89cdc7874..aaf390e44 100755
--- a/test/data-plane/library.sh
+++ b/test/data-plane/library.sh
@@ -41,7 +41,11 @@ 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}"
 
-readonly JAVA_IMAGE=docker.io/adoptopenjdk:14-jre-hotspot
+# 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
 
 readonly RECEIVER_JAR="receiver-1.0-SNAPSHOT.jar"
 readonly RECEIVER_DIRECTORY=receiver
@@ -87,6 +91,7 @@ function receiver_build_push() {
   docker build \
     -f ${DATA_PLANE_DIR}/docker/Dockerfile \
     --build-arg JAVA_IMAGE=${JAVA_IMAGE} \
+    --build-arg BASE_IMAGE=${BASE_IMAGE} \
     --build-arg APP_JAR=${RECEIVER_JAR} \
     --build-arg APP_DIR=${RECEIVER_DIRECTORY} \
     -t "${KNATIVE_KAFKA_BROKER_RECEIVER_IMAGE}" ${DATA_PLANE_DIR} &&
@@ -102,6 +107,7 @@ function dispatcher_build_push() {
   docker build \
     -f ${DATA_PLANE_DIR}/docker/Dockerfile \
     --build-arg JAVA_IMAGE=${JAVA_IMAGE} \
+    --build-arg BASE_IMAGE=${BASE_IMAGE} \
     --build-arg APP_JAR=${DISPATCHER_JAR} \
     --build-arg APP_DIR=${DISPATCHER_DIRECTORY} \
     -t "${KNATIVE_KAFKA_BROKER_DISPATCHER_IMAGE}" ${DATA_PLANE_DIR} &&
@@ -117,6 +123,7 @@ function sink_build_push() {
   docker build \
     -f ${DATA_PLANE_DIR}/docker/Dockerfile \
     --build-arg JAVA_IMAGE=${JAVA_IMAGE} \
+    --build-arg BASE_IMAGE=${BASE_IMAGE} \
     --build-arg APP_JAR=${RECEIVER_JAR} \
     --build-arg APP_DIR=${RECEIVER_DIRECTORY} \
     -t "${KNATIVE_KAFKA_SINK_RECEIVER_IMAGE}" ${DATA_PLANE_DIR} &&