Repository: knative-sandbox/eventing-kafka-broker
Pull request #71 (2020-07-31 13:49:43)

Add bootstrap servers as configuration
Bootstrap servers are a list of comma-separated host/port pairs to
use for establishing the initial connection to the Kafka cluster.
To change such a list, a user needs to change multiple ConfigMaps.
With these changes, a user can specify bootstrap servers in only one
ConfigMap, `kafka-broker-config`, and automatically propagate such
list from the control plane to the data plane through the usual
mechanism.

## Proposed Changes

- Add bootstrap servers as configuration

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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["mattmoor"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/71/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1289196676845670402/artifacts/line-cov.html#file0) | 90.4% | 91.1% | 0.8

knative-prow-robot: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | ca79988c99e07de22d6d0f7d89bf9d7df2b50ecf | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/71/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1289196676845670401) | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/71). [Your PR dashboard](https://gubernator.knative.dev/pr/pierDipi).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/71/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1289200830062268429/artifacts/line-cov.html#file0) | 90.4% | 91.1% | 0.8

slinkydeveloper: /lgtm
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/71#" title="Author self-approved">pierDipi</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/71#pullrequestreview-459790351" title="Approved">slinkydeveloper</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)~~ [slinkydeveloper]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow-robot: New changes are detected. LGTM label has been removed.
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/71/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1290184434712055809/artifacts/line-cov.html#file0) | 90.4% | 91.1% | 0.8

slinkydeveloper: /lgtm

Patch:
diff --git a/control-plane/pkg/core/config/triggers.pb.go b/control-plane/pkg/core/config/triggers.pb.go
index 14c6fa933..1181af92a 100644
--- a/control-plane/pkg/core/config/triggers.pb.go
+++ b/control-plane/pkg/core/config/triggers.pb.go
@@ -94,7 +94,9 @@ type Broker struct {
 	// broker namespace
 	Namespace string `protobuf:"bytes,5,opt,name=namespace,proto3" json:"namespace,omitempty"`
 	// broker name
-	Name                 string   `protobuf:"bytes,6,opt,name=name,proto3" json:"name,omitempty"`
+	Name string `protobuf:"bytes,6,opt,name=name,proto3" json:"name,omitempty"`
+	// A comma separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+	BootstrapServers     string   `protobuf:"bytes,7,opt,name=bootstrapServers,proto3" json:"bootstrapServers,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -167,6 +169,13 @@ func (m *Broker) GetName() string {
 	return ""
 }
 
+func (m *Broker) GetBootstrapServers() string {
+	if m != nil {
+		return m.BootstrapServers
+	}
+	return ""
+}
+
 type Brokers struct {
 	Brokers []*Broker `protobuf:"bytes,1,rep,name=brokers,proto3" json:"brokers,omitempty"`
 	// Count each config map update.
@@ -226,28 +235,29 @@ func init() {
 func init() { proto.RegisterFile("proto/def/triggers.proto", fileDescriptor_3cd32e421bcc2dd3) }
 
 var fileDescriptor_3cd32e421bcc2dd3 = []byte{
-	// 355 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x64, 0x92, 0x4f, 0x4b, 0xc3, 0x40,
-	0x10, 0xc5, 0x49, 0xfa, 0x27, 0xed, 0x14, 0x6b, 0x59, 0x3c, 0x2c, 0xa2, 0x50, 0x8b, 0x48, 0x11,
-	0xba, 0x01, 0xbd, 0x14, 0xc1, 0x83, 0x15, 0xf1, 0xe2, 0x29, 0x7a, 0x10, 0xc1, 0xc3, 0x36, 0x99,
-	0x86, 0x25, 0xe9, 0x6e, 0xd8, 0x6c, 0x03, 0xfd, 0x52, 0x9e, 0xfc, 0x80, 0x92, 0x4d, 0xd2, 0x96,
-	0x7a, 0x9b, 0xf9, 0xbd, 0xb7, 0x8f, 0xe1, 0xb1, 0x40, 0x33, 0xad, 0x8c, 0xf2, 0x23, 0x5c, 0xf9,
-	0x46, 0x8b, 0x38, 0x46, 0x9d, 0x33, 0x8b, 0x26, 0xbf, 0x0e, 0x78, 0x1f, 0x15, 0x22, 0x73, 0x00,
-	0x6e, 0x8c, 0x16, 0xcb, 0x8d, 0xc1, 0x9c, 0x3a, 0xe3, 0xd6, 0x74, 0x70, 0x47, 0x59, 0xad, 0xb2,
-	0xa7, 0x9d, 0xf4, 0x22, 0x8d, 0xde, 0x06, 0x07, 0x5e, 0x32, 0x86, 0x41, 0x84, 0xb9, 0x11, 0x92,
-	0x1b, 0xa1, 0x24, 0x75, 0xc7, 0xce, 0xb4, 0x1f, 0x1c, 0x22, 0x32, 0x04, 0x57, 0x44, 0xb4, 0x65,
-	0x05, 0x57, 0x44, 0xe7, 0x8f, 0x70, 0x7a, 0x14, 0x48, 0x46, 0xd0, 0x4a, 0x70, 0x4b, 0x1d, 0xeb,
-	0x29, 0x47, 0x72, 0x06, 0x9d, 0x82, 0xa7, 0x1b, 0xac, 0x03, 0xab, 0xe5, 0xc1, 0x9d, 0x3b, 0x93,
-	0x1f, 0x07, 0xba, 0x0b, 0xad, 0x12, 0xd4, 0x75, 0xb2, 0xd3, 0x24, 0x97, 0x8f, 0x8c, 0xca, 0x44,
-	0xd8, 0x3c, 0xb2, 0x0b, 0xb9, 0x81, 0x61, 0x84, 0x3c, 0x7a, 0x43, 0x63, 0x50, 0xbf, 0x0b, 0x99,
-	0xd4, 0xb7, 0x1c, 0x51, 0x72, 0x0d, 0xbd, 0xa6, 0x21, 0xda, 0xb6, 0x0d, 0xf4, 0x9a, 0x06, 0x82,
-	0x9d, 0x42, 0x2e, 0xa0, 0x2f, 0xf9, 0x1a, 0xf3, 0x8c, 0x87, 0x48, 0x3b, 0x36, 0x68, 0x0f, 0x08,
-	0x81, 0x76, 0xb9, 0xd0, 0xae, 0x15, 0xec, 0x3c, 0xf9, 0x04, 0xaf, 0xba, 0x37, 0x27, 0x57, 0xe0,
-	0x2d, 0xab, 0xb1, 0xee, 0xd8, 0x63, 0x95, 0x14, 0x34, 0x9c, 0xdc, 0xc2, 0xa8, 0x50, 0xe9, 0x66,
-	0x8d, 0xaf, 0x28, 0x51, 0xef, 0x4b, 0x6d, 0x07, 0xff, 0xf8, 0xe2, 0x1b, 0x66, 0x11, 0x16, 0x2c,
-	0x29, 0x8b, 0x2e, 0x90, 0x61, 0x81, 0xd2, 0x08, 0x19, 0xb3, 0x84, 0xaf, 0x12, 0xce, 0xaa, 0x44,
-	0x16, 0x2a, 0x8d, 0x2c, 0x54, 0x72, 0x25, 0xe2, 0xc5, 0x49, 0x7d, 0xc8, 0xb3, 0x5d, 0xbf, 0x2e,
-	0x43, 0x25, 0x8d, 0x56, 0xe9, 0x2c, 0x4b, 0xb9, 0x44, 0x3f, 0x4b, 0x62, 0xbf, 0x74, 0xfb, 0x95,
-	0x7b, 0xd9, 0xb5, 0xff, 0xe4, 0xfe, 0x2f, 0x00, 0x00, 0xff, 0xff, 0x25, 0x67, 0x41, 0x91, 0x43,
-	0x02, 0x00, 0x00,
+	// 377 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x64, 0x92, 0xcf, 0x6a, 0xe3, 0x30,
+	0x10, 0xc6, 0xb1, 0xf3, 0xc7, 0xc9, 0x84, 0xcd, 0x06, 0xb1, 0x07, 0xb1, 0xec, 0x42, 0x36, 0x2c,
+	0x25, 0x14, 0x22, 0x43, 0x7b, 0x09, 0x85, 0x1e, 0x9a, 0x52, 0x7a, 0xe9, 0xc9, 0xe9, 0xa1, 0x14,
+	0x7a, 0x90, 0xed, 0x89, 0x11, 0x76, 0x24, 0x23, 0x2b, 0x86, 0xbc, 0x57, 0x9f, 0xa6, 0x4f, 0x53,
+	0x2c, 0xdb, 0x49, 0x48, 0x6e, 0x33, 0xdf, 0xf7, 0xe9, 0xc7, 0x78, 0xc6, 0x40, 0x73, 0xad, 0x8c,
+	0xf2, 0x63, 0xdc, 0xf8, 0x46, 0x8b, 0x24, 0x41, 0x5d, 0x30, 0x2b, 0xcd, 0x3e, 0x1d, 0xf0, 0x5e,
+	0x6b, 0x89, 0x2c, 0x01, 0xb8, 0x31, 0x5a, 0x84, 0x3b, 0x83, 0x05, 0x75, 0xa6, 0x9d, 0xf9, 0xe8,
+	0x86, 0xb2, 0xc6, 0x65, 0x0f, 0x07, 0xeb, 0x49, 0x1a, 0xbd, 0x0f, 0x4e, 0xb2, 0x64, 0x0a, 0xa3,
+	0x18, 0x0b, 0x23, 0x24, 0x37, 0x42, 0x49, 0xea, 0x4e, 0x9d, 0xf9, 0x30, 0x38, 0x95, 0xc8, 0x18,
+	0x5c, 0x11, 0xd3, 0x8e, 0x35, 0x5c, 0x11, 0xff, 0xbe, 0x87, 0x9f, 0x67, 0x40, 0x32, 0x81, 0x4e,
+	0x8a, 0x7b, 0xea, 0xd8, 0x4c, 0x55, 0x92, 0x5f, 0xd0, 0x2b, 0x79, 0xb6, 0xc3, 0x06, 0x58, 0x37,
+	0x77, 0xee, 0xd2, 0x99, 0x7d, 0x39, 0xd0, 0x5f, 0x69, 0x95, 0xa2, 0x6e, 0xc8, 0x4e, 0x4b, 0xae,
+	0x1e, 0x19, 0x95, 0x8b, 0xa8, 0x7d, 0x64, 0x1b, 0x72, 0x05, 0xe3, 0x18, 0x79, 0xfc, 0x82, 0xc6,
+	0xa0, 0x5e, 0x0b, 0x99, 0x36, 0xb3, 0x9c, 0xa9, 0xe4, 0x3f, 0x0c, 0xda, 0x0d, 0xd1, 0xae, 0xdd,
+	0xc0, 0xa0, 0xdd, 0x40, 0x70, 0x70, 0xc8, 0x1f, 0x18, 0x4a, 0xbe, 0xc5, 0x22, 0xe7, 0x11, 0xd2,
+	0x9e, 0x05, 0x1d, 0x05, 0x42, 0xa0, 0x5b, 0x35, 0xb4, 0x6f, 0x0d, 0x5b, 0x93, 0x6b, 0x98, 0x84,
+	0x4a, 0x99, 0xc2, 0x68, 0x9e, 0xaf, 0x51, 0x97, 0x15, 0xdf, 0xb3, 0xfe, 0x85, 0x3e, 0x7b, 0x03,
+	0xaf, 0xfe, 0xb6, 0x82, 0xfc, 0x03, 0x2f, 0xac, 0xcb, 0xe6, 0x1e, 0x1e, 0xab, 0xad, 0xa0, 0xd5,
+	0x2b, 0x72, 0xa9, 0xb2, 0xdd, 0x16, 0x9f, 0x51, 0xa2, 0x3e, 0x1e, 0xa0, 0x1b, 0x5c, 0xe8, 0xab,
+	0x0f, 0x58, 0xc4, 0x58, 0xb2, 0xb4, 0x3a, 0x4a, 0x89, 0x0c, 0x4b, 0x94, 0x46, 0xc8, 0x84, 0xa5,
+	0x7c, 0x93, 0x72, 0x56, 0x13, 0x59, 0xa4, 0x34, 0xb2, 0x48, 0xc9, 0x8d, 0x48, 0x56, 0x3f, 0x9a,
+	0x41, 0x1e, 0x6d, 0xfb, 0xfe, 0x37, 0x52, 0xd2, 0x68, 0x95, 0x2d, 0xf2, 0x8c, 0x4b, 0xf4, 0xf3,
+	0x34, 0xf1, 0xab, 0xb4, 0x5f, 0xa7, 0xc3, 0xbe, 0xfd, 0xa7, 0x6e, 0xbf, 0x03, 0x00, 0x00, 0xff,
+	0xff, 0x80, 0x14, 0x6d, 0x63, 0x6f, 0x02, 0x00, 0x00,
 }
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index db9efc41b..b479a6f9c 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -246,11 +246,17 @@ func (r *Reconciler) topicDetailFromBrokerConfig(broker *eventing.Broker) *saram
 }
 
 func (r *Reconciler) getBrokerConfig(topic string, broker *eventing.Broker) (*coreconfig.Broker, error) {
+	bootstrapServers, err := r.getBootstrapServersOrFail()
+	if err != nil {
+		return nil, err
+	}
+
 	brokerConfig := &coreconfig.Broker{
-		Id:        string(broker.UID),
-		Topic:     topic,
-		Namespace: broker.Namespace,
-		Name:      broker.Name,
+		Id:               string(broker.UID),
+		Topic:            topic,
+		Namespace:        broker.Namespace,
+		Name:             broker.Name,
+		BootstrapServers: bootstrapServers,
 	}
 
 	if broker.Spec.Delivery == nil || broker.Spec.Delivery.DeadLetterSink == nil {
@@ -300,6 +306,10 @@ func (r *Reconciler) ConfigMapUpdated(ctx context.Context) func(configMap *corev
 // SetBootstrapServers change kafka bootstrap brokers addresses.
 // servers: a comma separated list of brokers to connect to.
 func (r *Reconciler) SetBootstrapServers(servers string) {
+	if servers == "" {
+		return
+	}
+
 	addrs := strings.Split(servers, ",")
 
 	r.bootstrapServersLock.Lock()
@@ -351,3 +361,14 @@ func deleteBroker(brokersTriggers *coreconfig.Brokers, index int) {
 	// truncate the array.
 	brokersTriggers.Brokers = brokersTriggers.Brokers[:len(brokersTriggers.Brokers)-1]
 }
+
+func (r *Reconciler) getBootstrapServersOrFail() (string, error) {
+	r.bootstrapServersLock.RLock()
+	defer r.bootstrapServersLock.RUnlock()
+
+	if len(r.bootstrapServers) == 0 {
+		return "", fmt.Errorf("no %s provided", BootstrapServersConfigMapKey)
+	}
+
+	return strings.Join(r.bootstrapServers, ","), nil
+}
diff --git a/data-plane/config/100-config-kafka-broker-data-plane.yaml b/data-plane/config/100-config-kafka-broker-data-plane.yaml
index 320ce9ce4..18622c68f 100644
--- a/data-plane/config/100-config-kafka-broker-data-plane.yaml
+++ b/data-plane/config/100-config-kafka-broker-data-plane.yaml
@@ -21,7 +21,6 @@ metadata:
   namespace: knative-eventing
 data:
   config-kafka-broker-producer.properties: |
-    bootstrap.servers=my-cluster-kafka-bootstrap.kafka:9092
     acks=1
     buffer.memory=33554432
     # compression.type=snappy
@@ -51,7 +50,6 @@ data:
     # transaction.timeout.ms=60000
     # transactional.id=null
   config-kafka-broker-consumer.properties: |
-    bootstrap.servers=my-cluster-kafka-bootstrap.kafka:9092
     fetch.min.bytes=1
     heartbeat.interval.ms=3000
     max.partition.fetch.bytes=1048576
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/Broker.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/Broker.java
index 805552ffa..9128ae23d 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/Broker.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/Broker.java
@@ -62,4 +62,19 @@ public interface Broker {
    * @return name.
    */
   String name();
+
+  /**
+   * A comma separated list of host/port pairs to use for establishing the initial connection to the
+   * Kafka cluster.
+   *
+   * @return bootstrap servers.
+   */
+  String bootstrapServers();
+
+  /**
+   * Get request path to accept events for this Broker.
+   *
+   * @return request path associated with this Broker.
+   */
+  String path();
 }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/BrokerWrapper.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/BrokerWrapper.java
index b23ae27e0..3346b097e 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/BrokerWrapper.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/BrokerWrapper.java
@@ -62,6 +62,16 @@ public class BrokerWrapper implements dev.knative.eventing.kafka.broker.core.Bro
     return broker.getName();
   }
 
+  @Override
+  public String bootstrapServers() {
+    return broker.getBootstrapServers();
+  }
+
+  @Override
+  public String path() {
+    return "/" + namespace() + "/" + name();
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -76,7 +86,9 @@ public class BrokerWrapper implements dev.knative.eventing.kafka.broker.core.Bro
         && broker.getDeadLetterSink().equals(that.deadLetterSink())
         && broker.getTopic().equals(that.topic())
         && broker.getName().equals(that.name())
-        && broker.getNamespace().equals(that.namespace());
+        && broker.getNamespace().equals(that.namespace())
+        && broker.getBootstrapServers().equals(that.bootstrapServers())
+        && path().equals(that.path());
   }
 
   @Override
@@ -86,7 +98,9 @@ public class BrokerWrapper implements dev.knative.eventing.kafka.broker.core.Bro
         broker.getDeadLetterSink(),
         broker.getTopic(),
         broker.getNamespace(),
-        broker.getName()
+        broker.getName(),
+        broker.getBootstrapServers(),
+        path()
     );
   }
 
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 01d7c811b..b101c5c3b 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
@@ -38,8 +38,10 @@ import io.vertx.core.http.HttpClientResponse;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import java.util.Objects;
 import java.util.Properties;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 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;
 
@@ -144,6 +146,7 @@ public class HttpConsumerVerticleFactory implements ConsumerVerticleFactory<Clou
     // TODO check producer configurations to change per instance
     // producerConfigs is a shared object and it acts as a prototype for each consumer instance.
     final var producerConfigs = (Properties) this.producerConfigs.clone();
+    producerConfigs.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.bootstrapServers());
 
     final var kafkaProducer = new KafkaProducer<>(
         producerConfigs,
@@ -163,6 +166,7 @@ public class HttpConsumerVerticleFactory implements ConsumerVerticleFactory<Clou
     // consumerConfigs is a shared object and it acts as a prototype for each consumer instance.
     final var consumerConfigs = (Properties) this.consumerConfigs.clone();
     consumerConfigs.setProperty(GROUP_ID_CONFIG, trigger.id());
+    consumerConfigs.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.bootstrapServers());
 
     // Note: KafkaConsumer instances are not thread-safe.
     // There are methods thread-safe, but in general they're not.
diff --git a/data-plane/generated/src/main/java/dev/knative/eventing/kafka/broker/core/config/BrokersConfig.java b/data-plane/generated/src/main/java/dev/knative/eventing/kafka/broker/core/config/BrokersConfig.java
index b23fa0e5b..119ccb43f 100644
--- a/data-plane/generated/src/main/java/dev/knative/eventing/kafka/broker/core/config/BrokersConfig.java
+++ b/data-plane/generated/src/main/java/dev/knative/eventing/kafka/broker/core/config/BrokersConfig.java
@@ -1345,6 +1345,24 @@ public final class BrokersConfig {
      */
     com.google.protobuf.ByteString
         getNameBytes();
+
+    /**
+     * <pre>
+     * A comma separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+     * </pre>
+     *
+     * <code>string bootstrapServers = 7;</code>
+     */
+    java.lang.String getBootstrapServers();
+    /**
+     * <pre>
+     * A comma separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+     * </pre>
+     *
+     * <code>string bootstrapServers = 7;</code>
+     */
+    com.google.protobuf.ByteString
+        getBootstrapServersBytes();
   }
   /**
    * Protobuf type {@code Broker}
@@ -1365,6 +1383,7 @@ public final class BrokersConfig {
       triggers_ = java.util.Collections.emptyList();
       namespace_ = "";
       name_ = "";
+      bootstrapServers_ = "";
     }
 
     @java.lang.Override
@@ -1437,6 +1456,12 @@ public final class BrokersConfig {
               name_ = s;
               break;
             }
+            case 58: {
+              java.lang.String s = input.readStringRequireUtf8();
+
+              bootstrapServers_ = s;
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -1729,6 +1754,48 @@ public final class BrokersConfig {
       }
     }
 
+    public static final int BOOTSTRAPSERVERS_FIELD_NUMBER = 7;
+    private volatile java.lang.Object bootstrapServers_;
+    /**
+     * <pre>
+     * A comma separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+     * </pre>
+     *
+     * <code>string bootstrapServers = 7;</code>
+     */
+    public java.lang.String getBootstrapServers() {
+      java.lang.Object ref = bootstrapServers_;
+      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();
+        bootstrapServers_ = s;
+        return s;
+      }
+    }
+    /**
+     * <pre>
+     * A comma separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+     * </pre>
+     *
+     * <code>string bootstrapServers = 7;</code>
+     */
+    public com.google.protobuf.ByteString
+        getBootstrapServersBytes() {
+      java.lang.Object ref = bootstrapServers_;
+      if (ref instanceof java.lang.String) {
+        com.google.protobuf.ByteString b = 
+            com.google.protobuf.ByteString.copyFromUtf8(
+                (java.lang.String) ref);
+        bootstrapServers_ = b;
+        return b;
+      } else {
+        return (com.google.protobuf.ByteString) ref;
+      }
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -1761,6 +1828,9 @@ public final class BrokersConfig {
       if (!getNameBytes().isEmpty()) {
         com.google.protobuf.GeneratedMessageV3.writeString(output, 6, name_);
       }
+      if (!getBootstrapServersBytes().isEmpty()) {
+        com.google.protobuf.GeneratedMessageV3.writeString(output, 7, bootstrapServers_);
+      }
       unknownFields.writeTo(output);
     }
 
@@ -1789,6 +1859,9 @@ public final class BrokersConfig {
       if (!getNameBytes().isEmpty()) {
         size += com.google.protobuf.GeneratedMessageV3.computeStringSize(6, name_);
       }
+      if (!getBootstrapServersBytes().isEmpty()) {
+        size += com.google.protobuf.GeneratedMessageV3.computeStringSize(7, bootstrapServers_);
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -1816,6 +1889,8 @@ public final class BrokersConfig {
           .equals(other.getNamespace())) return false;
       if (!getName()
           .equals(other.getName())) return false;
+      if (!getBootstrapServers()
+          .equals(other.getBootstrapServers())) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -1841,6 +1916,8 @@ public final class BrokersConfig {
       hash = (53 * hash) + getNamespace().hashCode();
       hash = (37 * hash) + NAME_FIELD_NUMBER;
       hash = (53 * hash) + getName().hashCode();
+      hash = (37 * hash) + BOOTSTRAPSERVERS_FIELD_NUMBER;
+      hash = (53 * hash) + getBootstrapServers().hashCode();
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -1991,6 +2068,8 @@ public final class BrokersConfig {
 
         name_ = "";
 
+        bootstrapServers_ = "";
+
         return this;
       }
 
@@ -2032,6 +2111,7 @@ public final class BrokersConfig {
         }
         result.namespace_ = namespace_;
         result.name_ = name_;
+        result.bootstrapServers_ = bootstrapServers_;
         onBuilt();
         return result;
       }
@@ -2126,6 +2206,10 @@ public final class BrokersConfig {
           name_ = other.name_;
           onChanged();
         }
+        if (!other.getBootstrapServers().isEmpty()) {
+          bootstrapServers_ = other.bootstrapServers_;
+          onChanged();
+        }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -2892,6 +2976,95 @@ public final class BrokersConfig {
         onChanged();
         return this;
       }
+
+      private java.lang.Object bootstrapServers_ = "";
+      /**
+       * <pre>
+       * A comma separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+       * </pre>
+       *
+       * <code>string bootstrapServers = 7;</code>
+       */
+      public java.lang.String getBootstrapServers() {
+        java.lang.Object ref = bootstrapServers_;
+        if (!(ref instanceof java.lang.String)) {
+          com.google.protobuf.ByteString bs =
+              (com.google.protobuf.ByteString) ref;
+          java.lang.String s = bs.toStringUtf8();
+          bootstrapServers_ = s;
+          return s;
+        } else {
+          return (java.lang.String) ref;
+        }
+      }
+      /**
+       * <pre>
+       * A comma separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+       * </pre>
+       *
+       * <code>string bootstrapServers = 7;</code>
+       */
+      public com.google.protobuf.ByteString
+          getBootstrapServersBytes() {
+        java.lang.Object ref = bootstrapServers_;
+        if (ref instanceof String) {
+          com.google.protobuf.ByteString b = 
+              com.google.protobuf.ByteString.copyFromUtf8(
+                  (java.lang.String) ref);
+          bootstrapServers_ = b;
+          return b;
+        } else {
+          return (com.google.protobuf.ByteString) ref;
+        }
+      }
+      /**
+       * <pre>
+       * A comma separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+       * </pre>
+       *
+       * <code>string bootstrapServers = 7;</code>
+       */
+      public Builder setBootstrapServers(
+          java.lang.String value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  
+        bootstrapServers_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <pre>
+       * A comma separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+       * </pre>
+       *
+       * <code>string bootstrapServers = 7;</code>
+       */
+      public Builder clearBootstrapServers() {
+        
+        bootstrapServers_ = getDefaultInstance().getBootstrapServers();
+        onChanged();
+        return this;
+      }
+      /**
+       * <pre>
+       * A comma separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+       * </pre>
+       *
+       * <code>string bootstrapServers = 7;</code>
+       */
+      public Builder setBootstrapServersBytes(
+          com.google.protobuf.ByteString value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  checkByteStringIsUtf8(value);
+        
+        bootstrapServers_ = value;
+        onChanged();
+        return this;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final com.google.protobuf.UnknownFieldSet unknownFields) {
@@ -3854,14 +4027,15 @@ public final class BrokersConfig {
       "\n\nattributes\030\001 \003(\0132\030.Trigger.AttributesE" +
       "ntry\022\023\n\013destination\030\002 \001(\t\022\n\n\002id\030\003 \001(\t\0321\n" +
       "\017AttributesEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002" +
-      " \001(\t:\0028\001\"x\n\006Broker\022\n\n\002id\030\001 \001(\t\022\r\n\005topic\030" +
-      "\002 \001(\t\022\026\n\016deadLetterSink\030\003 \001(\t\022\032\n\010trigger" +
-      "s\030\004 \003(\0132\010.Trigger\022\021\n\tnamespace\030\005 \001(\t\022\014\n\004" +
-      "name\030\006 \001(\t\"=\n\007Brokers\022\030\n\007brokers\030\001 \003(\0132\007" +
-      ".Broker\022\030\n\020volumeGeneration\030\002 \001(\004B]\n-dev" +
-      ".knative.eventing.kafka.broker.core.conf" +
-      "igB\rBrokersConfigZ\035control-plane/pkg/cor" +
-      "e/configb\006proto3"
+      " \001(\t:\0028\001\"\222\001\n\006Broker\022\n\n\002id\030\001 \001(\t\022\r\n\005topic" +
+      "\030\002 \001(\t\022\026\n\016deadLetterSink\030\003 \001(\t\022\032\n\010trigge" +
+      "rs\030\004 \003(\0132\010.Trigger\022\021\n\tnamespace\030\005 \001(\t\022\014\n" +
+      "\004name\030\006 \001(\t\022\030\n\020bootstrapServers\030\007 \001(\t\"=\n" +
+      "\007Brokers\022\030\n\007brokers\030\001 \003(\0132\007.Broker\022\030\n\020vo" +
+      "lumeGeneration\030\002 \001(\004B]\n-dev.knative.even" +
+      "ting.kafka.broker.core.configB\rBrokersCo" +
+      "nfigZ\035control-plane/pkg/core/configb\006pro" +
+      "to3"
     };
     descriptor = com.google.protobuf.Descriptors.FileDescriptor
       .internalBuildGeneratedFileFrom(descriptorData,
@@ -3884,7 +4058,7 @@ public final class BrokersConfig {
     internal_static_Broker_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Broker_descriptor,
-        new java.lang.String[] { "Id", "Topic", "DeadLetterSink", "Triggers", "Namespace", "Name", });
+        new java.lang.String[] { "Id", "Topic", "DeadLetterSink", "Triggers", "Namespace", "Name", "BootstrapServers", });
     internal_static_Brokers_descriptor =
       getDescriptor().getMessageTypes().get(2);
     internal_static_Brokers_fieldAccessorTable = new
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/Main.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/Main.java
index 5b3ce7080..bc89fe5b5 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/Main.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/Main.java
@@ -16,7 +16,6 @@
 
 package dev.knative.eventing.kafka.broker.receiver;
 
-import static io.vertx.kafka.client.producer.KafkaProducer.createShared;
 import static net.logstash.logback.argument.StructuredArguments.keyValue;
 
 import dev.knative.eventing.kafka.broker.core.ObjectsCreator;
@@ -24,6 +23,7 @@ import dev.knative.eventing.kafka.broker.core.file.FileWatcher;
 import io.cloudevents.kafka.CloudEventSerializer;
 import io.vertx.core.Vertx;
 import io.vertx.core.http.HttpServerOptions;
+import io.vertx.kafka.client.producer.KafkaProducer;
 import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
@@ -69,15 +69,12 @@ public class Main {
     final var vertx = Vertx.vertx();
     producerConfigs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CloudEventSerializer.class);
     producerConfigs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-    final var producer = createShared(
-        vertx,
-        PRODUCER_NAME,
+
+    final var handler = new RequestHandler<>(
         producerConfigs,
-        new StringSerializer(),
-        new CloudEventSerializer()
+        new CloudEventRequestToRecordMapper(),
+        properties -> KafkaProducer.create(vertx, properties)
     );
-
-    final var handler = new RequestHandler<>(producer, new CloudEventRequestToRecordMapper());
     final var httpServerOptions = new HttpServerOptions();
     httpServerOptions.setPort(env.getIngressPort());
     final var verticle = new HttpVerticle(httpServerOptions, new SimpleProbeHandlerDecorator(
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestHandler.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestHandler.java
index 456f83ad9..d782fc6de 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestHandler.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestHandler.java
@@ -33,21 +33,22 @@ import io.vertx.core.http.HttpServerRequest;
 import io.vertx.kafka.client.producer.KafkaProducer;
 import io.vertx.kafka.client.producer.KafkaProducerRecord;
 import io.vertx.kafka.client.producer.RecordMetadata;
-import java.util.HashSet;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Objects;
+import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Collectors;
+import java.util.function.Function;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * RequestHandler is responsible for mapping HTTP requests to Kafka records, sending records to
  * Kafka through the Kafka producer and terminating requests with the appropriate status code.
- *
- * @param <K> type of the records' key.
- * @param <V> type of the records' value.
  */
 public class RequestHandler<K, V> implements Handler<HttpServerRequest>,
     ObjectsReconciler<CloudEvent> {
@@ -59,37 +60,43 @@ public class RequestHandler<K, V> implements Handler<HttpServerRequest>,
 
   private static final Logger logger = LoggerFactory.getLogger(RequestHandler.class);
 
-  private final KafkaProducer<K, V> producer;
   private final RequestToRecordMapper<K, V> requestToRecordMapper;
-  private final AtomicReference<Set<String>> brokers;
+  // path -> <bootstrapServers, producer>
+  private final AtomicReference<Map<String, Entry<String, KafkaProducer<K, V>>>> producers;
+  private final Properties producerConfigs;
+  private final Function<Properties, KafkaProducer<K, V>> producerCreator;
 
   /**
    * Create a new Request handler.
    *
-   * @param producer              kafka producer
+   * @param producerConfigs       common producers configurations
    * @param requestToRecordMapper request to record mapper
    */
   public RequestHandler(
-      final KafkaProducer<K, V> producer,
-      final RequestToRecordMapper<K, V> requestToRecordMapper) {
+      final Properties producerConfigs,
+      final RequestToRecordMapper<K, V> requestToRecordMapper,
+      final Function<Properties, KafkaProducer<K, V>> producerCreator) {
 
-    Objects.requireNonNull(producer, "provide a producer");
+    Objects.requireNonNull(producerConfigs, "provide producerConfigs");
     Objects.requireNonNull(requestToRecordMapper, "provide a mapper");
+    Objects.requireNonNull(producerCreator, "provide producerCreator");
 
-    this.producer = producer;
+    this.producerConfigs = producerConfigs;
     this.requestToRecordMapper = requestToRecordMapper;
-    brokers = new AtomicReference<>(new HashSet<>());
+    this.producerCreator = producerCreator;
+    producers = new AtomicReference<>(new HashMap<>());
   }
 
   @Override
   public void handle(final HttpServerRequest request) {
 
-    if (!brokers.get().contains(request.path())) {
+    final var producer = producers.get().get(request.path());
+    if (producer == null) {
 
       request.response().setStatusCode(BROKER_NOT_FOUND).end();
 
       logger.warn("broker not found {} {}",
-          keyValue("brokers", brokers.get()),
+          keyValue("brokers", producers.get().keySet()),
           keyValue("path", request.path())
       );
 
@@ -98,7 +105,7 @@ public class RequestHandler<K, V> implements Handler<HttpServerRequest>,
 
     requestToRecordMapper
         .recordFromRequest(request)
-        .onSuccess(record -> send(record)
+        .onSuccess(record -> send(producer.getValue(), record)
             .onSuccess(ignore -> {
               request.response().setStatusCode(RECORD_PRODUCED).end();
 
@@ -127,7 +134,10 @@ public class RequestHandler<K, V> implements Handler<HttpServerRequest>,
         });
   }
 
-  private Future<RecordMetadata> send(final KafkaProducerRecord<K, V> record) {
+  private static <K, V> Future<RecordMetadata> send(
+      final KafkaProducer<K, V> producer,
+      final KafkaProducerRecord<K, V> record) {
+
     final Promise<RecordMetadata> promise = Promise.promise();
     producer.send(record, promise);
     return promise.future();
@@ -136,14 +146,49 @@ public class RequestHandler<K, V> implements Handler<HttpServerRequest>,
   @Override
   public Future<Void> reconcile(Map<Broker, Set<Trigger<CloudEvent>>> objects) {
 
-    final var brokers = objects.keySet().stream()
-        .map(b -> "/" + b.namespace() + "/" + b.name())
-        .collect(Collectors.toSet());
+    final Map<String, Entry<String, KafkaProducer<K, V>>> newProducers
+        = new HashMap<>();
+
+    final var producers = this.producers.get();
+
+    for (final var broker : objects.keySet()) {
+      final var pair = producers.get(broker.path());
+
+      if (pair == null) {
+        // There is no producer for this Broker, so create it and add it to newProducers.
+        addBroker(newProducers, broker);
+        continue;
+      }
+
+      if (!pair.getKey().equals(broker.bootstrapServers())) {
+        // Bootstrap servers changed, close the old producer, and re-create a new one.
+        final var producer = pair.getValue();
+        producer.flush(complete -> producer.close());
 
-    this.brokers.set(brokers);
+        addBroker(newProducers, broker);
+        continue;
+      }
 
-    logger.debug("Added brokers to handler {}", keyValue("brokers", brokers));
+      // Nothing changed, so add the previous entry, to newProducers.
+      newProducers.put(broker.path(), pair);
+    }
+
+    this.producers.set(newProducers);
+
+    logger.debug("Added brokers to handler {}", keyValue("brokers", newProducers.keySet()));
 
     return Future.succeededFuture();
   }
+
+  private void addBroker(
+      final Map<String, Entry<String, KafkaProducer<K, V>>> producers,
+      final Broker broker) {
+
+    final var producerConfigs = (Properties) this.producerConfigs.clone();
+    producerConfigs.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.bootstrapServers());
+
+    final KafkaProducer<K, V> producer = producerCreator.apply(producerConfigs);
+
+    producers.put(broker.path(), new SimpleImmutableEntry<>(broker.bootstrapServers(), producer));
+  }
 }
diff --git a/proto/def/triggers.proto b/proto/def/triggers.proto
index 8898946b4..66533df55 100644
--- a/proto/def/triggers.proto
+++ b/proto/def/triggers.proto
@@ -39,6 +39,9 @@ message Broker {
 
   // broker name
   string name = 6;
+
+ // A comma separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
+  string bootstrapServers = 7;
 }
 
 message Brokers {


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index d5b3d352e..ce3854c33 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -59,6 +59,8 @@ const (
 
 const (
 	finalizerName = "brokers.eventing.knative.dev"
+
+	bootstrapServers = "kafka-1:9092,kafka-2:9093"
 )
 
 var (
@@ -112,10 +114,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				ConfigMapUpdate(&configs, &coreconfig.Brokers{
 					Brokers: []*coreconfig.Broker{
 						{
-							Id:        BrokerUUID,
-							Topic:     GetTopic(),
-							Namespace: BrokerNamespace,
-							Name:      BrokerName,
+							Id:               BrokerUUID,
+							Topic:            GetTopic(),
+							Namespace:        BrokerNamespace,
+							Name:             BrokerName,
+							BootstrapServers: bootstrapServers,
 						},
 					},
 					VolumeGeneration: 1,
@@ -142,6 +145,9 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					),
 				},
 			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
 		},
 		{
 			Name: "Reconciled normal - with DLS",
@@ -164,11 +170,12 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				ConfigMapUpdate(&configs, &coreconfig.Brokers{
 					Brokers: []*coreconfig.Broker{
 						{
-							Id:             BrokerUUID,
-							Topic:          GetTopic(),
-							DeadLetterSink: "http://test-service.test-service-namespace.svc.cluster.local/",
-							Namespace:      BrokerNamespace,
-							Name:           BrokerName,
+							Id:               BrokerUUID,
+							Topic:            GetTopic(),
+							DeadLetterSink:   "http://test-service.test-service-namespace.svc.cluster.local/",
+							Namespace:        BrokerNamespace,
+							Name:             BrokerName,
+							BootstrapServers: bootstrapServers,
 						},
 					},
 					VolumeGeneration: 2,
@@ -194,6 +201,9 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					),
 				},
 			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
 		},
 		{
 			Name: "Failed to create topic",
@@ -254,11 +264,12 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				ConfigMapUpdate(&configs, &coreconfig.Brokers{
 					Brokers: []*coreconfig.Broker{
 						{
-							Id:             BrokerUUID,
-							Topic:          GetTopic(),
-							DeadLetterSink: "http://test-service.test-service-namespace.svc.cluster.local/",
-							Namespace:      BrokerNamespace,
-							Name:           BrokerName,
+							Id:               BrokerUUID,
+							Topic:            GetTopic(),
+							DeadLetterSink:   "http://test-service.test-service-namespace.svc.cluster.local/",
+							Namespace:        BrokerNamespace,
+							Name:             BrokerName,
+							BootstrapServers: bootstrapServers,
 						},
 					},
 					VolumeGeneration: 1,
@@ -284,6 +295,9 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					),
 				},
 			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
 		},
 		{
 			Name: "Reconciled normal - config map not readable",
@@ -304,10 +318,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				ConfigMapUpdate(&configs, &coreconfig.Brokers{
 					Brokers: []*coreconfig.Broker{
 						{
-							Id:        BrokerUUID,
-							Topic:     GetTopic(),
-							Namespace: BrokerNamespace,
-							Name:      BrokerName,
+							Id:               BrokerUUID,
+							Topic:            GetTopic(),
+							Namespace:        BrokerNamespace,
+							Name:             BrokerName,
+							BootstrapServers: bootstrapServers,
 						},
 					},
 					VolumeGeneration: 1,
@@ -332,6 +347,9 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					),
 				},
 			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
 		},
 		{
 			Name: "Reconciled normal - preserve config map previous state",
@@ -381,10 +399,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							DeadLetterSink: "http://www.my-sink.com",
 						},
 						{
-							Id:        BrokerUUID,
-							Topic:     GetTopic(),
-							Namespace: BrokerNamespace,
-							Name:      BrokerName,
+							Id:               BrokerUUID,
+							Topic:            GetTopic(),
+							Namespace:        BrokerNamespace,
+							Name:             BrokerName,
+							BootstrapServers: bootstrapServers,
 						},
 					},
 					VolumeGeneration: 1,
@@ -409,6 +428,9 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					),
 				},
 			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
 		},
 		{
 			Name: "Reconciled normal - update existing broker while preserving others",
@@ -461,11 +483,12 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							DeadLetterSink: "http://www.my-sink.com",
 						},
 						{
-							Id:             BrokerUUID,
-							Topic:          GetTopic(),
-							DeadLetterSink: "http://www.my-sink.com/api",
-							Namespace:      BrokerNamespace,
-							Name:           BrokerName,
+							Id:               BrokerUUID,
+							Topic:            GetTopic(),
+							DeadLetterSink:   "http://www.my-sink.com/api",
+							Namespace:        BrokerNamespace,
+							Name:             BrokerName,
+							BootstrapServers: bootstrapServers,
 						},
 					},
 					VolumeGeneration: 1,
@@ -500,6 +523,9 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					),
 				},
 			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
 		},
 		{
 			Name: "Reconciled normal - remove existing broker DLS while preserving others",
@@ -550,10 +576,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							Name:           BrokerName,
 						},
 						{
-							Id:        BrokerUUID,
-							Topic:     GetTopic(),
-							Namespace: BrokerNamespace,
-							Name:      BrokerName,
+							Id:               BrokerUUID,
+							Topic:            GetTopic(),
+							Namespace:        BrokerNamespace,
+							Name:             BrokerName,
+							BootstrapServers: bootstrapServers,
 						},
 					},
 					VolumeGeneration: 1,
@@ -581,6 +608,9 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					),
 				},
 			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
 		},
 		{
 			Name: "Reconciled normal - increment volume generation",
@@ -627,10 +657,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							Name:           BrokerName,
 						},
 						{
-							Id:        BrokerUUID,
-							Topic:     GetTopic(),
-							Namespace: BrokerNamespace,
-							Name:      BrokerName,
+							Id:               BrokerUUID,
+							Topic:            GetTopic(),
+							Namespace:        BrokerNamespace,
+							Name:             BrokerName,
+							BootstrapServers: bootstrapServers,
 						},
 					},
 					VolumeGeneration: 2,
@@ -655,6 +686,9 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					),
 				},
 			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
 		},
 		{
 			Name: "Failed to resolve DLS",
@@ -676,10 +710,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							Name:           BrokerName,
 						},
 						{
-							Id:        BrokerUUID,
-							Topic:     GetTopic(),
-							Namespace: BrokerNamespace,
-							Name:      BrokerName,
+							Id:               BrokerUUID,
+							Topic:            GetTopic(),
+							Namespace:        BrokerNamespace,
+							Name:             BrokerName,
+							BootstrapServers: bootstrapServers,
 						},
 					},
 					VolumeGeneration: 1,
@@ -718,6 +753,40 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					),
 				},
 			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
+		{
+			Name: "No bootstrap.servers provided",
+			Objects: []runtime.Object{
+				NewBroker(),
+				NewConfigMapFromBrokers(&coreconfig.Brokers{
+					Brokers:          []*coreconfig.Broker{},
+					VolumeGeneration: 1,
+				}, &configs),
+			},
+			Key:     testKey,
+			WantErr: true,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					"failed to get broker configuration: no bootstrap.servers provided",
+				),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						reconcilertesting.WithInitBrokerConditions,
+						TopicReady,
+					),
+				},
+			},
 		},
 	}
 
@@ -978,6 +1047,11 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 			onDeleteTopicError = want.(error)
 		}
 
+		bootstrapServers := ""
+		if bs, ok := row.OtherTestData[BootstrapServersConfigMapKey]; ok {
+			bootstrapServers = bs.(string)
+		}
+
 		reconciler := &Reconciler{
 			Reconciler: &base.Reconciler{
 				KubeClient:                  kubeclient.Get(ctx),
@@ -1000,6 +1074,7 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 			KafkaDefaultTopicDetailsLock: sync.RWMutex{},
 			Configs:                      configs,
 		}
+		reconciler.SetBootstrapServers(bootstrapServers)
 
 		r := brokerreconciler.NewReconciler(
 			ctx,
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/BrokerWrapperTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/BrokerWrapperTest.java
index 78edea536..ad32b201b 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/BrokerWrapperTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/BrokerWrapperTest.java
@@ -109,6 +109,18 @@ public class BrokerWrapperTest {
             new BrokerWrapper(
                 Broker.newBuilder().build()
             )
+        ),
+        Arguments.of(
+            new BrokerWrapper(
+                Broker.newBuilder()
+                    .setBootstrapServers("kafka-1:9092,kafka-2:9093")
+                    .build()
+            ),
+            new BrokerWrapper(
+                Broker.newBuilder()
+                    .setBootstrapServers("kafka-1:9092,kafka-2:9092")
+                    .build()
+            )
         )
     );
   }
@@ -187,6 +199,18 @@ public class BrokerWrapperTest {
                     .setDeadLetterSink("http://localhost:9090")
                     .build()
             )
+        ),
+        Arguments.of(
+            new BrokerWrapper(
+                Broker.newBuilder()
+                    .setBootstrapServers("kafka-1:9092,kafka-2:9092")
+                    .build()
+            ),
+            new BrokerWrapper(
+                Broker.newBuilder()
+                    .setBootstrapServers("kafka-1:9092,kafka-2:9092")
+                    .build()
+            )
         )
     );
   }
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 a7bb9c299..93d0efd64 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
@@ -94,6 +94,16 @@ public class HttpConsumerVerticleFactoryTest {
           public String name() {
             return null;
           }
+
+          @Override
+          public String bootstrapServers() {
+            return "0.0.0.0:9092";
+          }
+
+          @Override
+          public String path() {
+            return null;
+          }
         },
         new Trigger<>() {
           @Override
@@ -116,6 +126,7 @@ public class HttpConsumerVerticleFactoryTest {
     assertThat(consumerFactoryFuture.succeeded()).isTrue();
   }
 
+  @Test
   public void shouldNotThrowIllegalArgumentExceptionIfNotDLQ(final Vertx vertx) {
 
     final var consumerProperties = new Properties();
@@ -166,6 +177,16 @@ public class HttpConsumerVerticleFactoryTest {
             public String name() {
               return null;
             }
+
+            @Override
+            public String bootstrapServers() {
+              return "0.0.0.0:9092";
+            }
+
+            @Override
+            public String path() {
+              return null;
+            }
           },
           new Trigger<>() {
             @Override
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/RequestHandlerTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/RequestHandlerTest.java
index f317250b9..1faeb60bd 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/RequestHandlerTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/RequestHandlerTest.java
@@ -17,6 +17,7 @@
 package dev.knative.eventing.kafka.broker.receiver;
 
 import static dev.knative.eventing.kafka.broker.core.testing.utils.CoreObjects.broker1;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
@@ -24,20 +25,28 @@ import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import dev.knative.eventing.kafka.broker.core.BrokerWrapper;
+import dev.knative.eventing.kafka.broker.core.config.BrokersConfig.Broker;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.http.HttpServerRequest;
 import io.vertx.core.http.HttpServerResponse;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
 import io.vertx.kafka.client.producer.KafkaProducer;
 import io.vertx.kafka.client.producer.RecordMetadata;
 import io.vertx.kafka.client.producer.impl.KafkaProducerRecordImpl;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
+@ExtendWith(VertxExtension.class)
 public class RequestHandlerTest {
 
   private static final int TIMEOUT = 3;
@@ -80,10 +89,14 @@ public class RequestHandlerTest {
     final var broker = broker1();
 
     final var request = mock(HttpServerRequest.class);
-    when(request.path()).thenReturn(String.format("/%s/%s", broker.namespace(), broker.name()));
+    when(request.path()).thenReturn(broker.path());
     final var response = mockResponse(request, statusCode);
 
-    final var handler = new RequestHandler<>(producer, mapper);
+    final var handler = new RequestHandler<>(
+        new Properties(),
+        mapper,
+        properties -> producer
+    );
 
     final var countDown = new CountDownLatch(1);
 
@@ -112,7 +125,11 @@ public class RequestHandlerTest {
     when(request.path()).thenReturn(String.format("/%s/%s", broker.namespace(), broker.name()));
     final var response = mockResponse(request, RequestHandler.MAPPER_FAILED);
 
-    final var handler = new RequestHandler<Object, Object>(producer, mapper);
+    final var handler = new RequestHandler<Object, Object>(
+        new Properties(),
+        mapper,
+        properties -> producer
+    );
 
     final var countDown = new CountDownLatch(1);
     handler.reconcile(Map.of(broker, new HashSet<>()))
@@ -143,4 +160,93 @@ public class RequestHandlerTest {
     when(request.response()).thenReturn(response);
     return response;
   }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  public void shouldRecreateProducerWhenBootstrapServerChange(final VertxTestContext context) {
+
+    final RequestToRecordMapper<Object, Object> mapper
+        = (request) -> Future.succeededFuture();
+
+    final var first = new AtomicBoolean(true);
+    final var recreated = new AtomicBoolean(false);
+
+    final var handler = new RequestHandler<Object, Object>(
+        new Properties(),
+        mapper,
+        properties -> {
+          if (!first.getAndSet(false)) {
+            recreated.set(true);
+          }
+          return mock(KafkaProducer.class);
+        }
+    );
+
+    final var checkpoint = context.checkpoint();
+
+    final var broker1 = new BrokerWrapper(Broker.newBuilder()
+        .setId("1")
+        .setBootstrapServers("kafka-1:9092,kafka-2:9092")
+        .build());
+
+    final var broker2 = new BrokerWrapper(Broker.newBuilder()
+        .setId("1")
+        .setBootstrapServers("kafka-1:9092,kafka-3:9092")
+        .build());
+
+    handler.reconcile(Map.of(broker1, new HashSet<>()))
+        .onSuccess(ignored -> handler.reconcile(Map.of(broker2, new HashSet<>()))
+            .onSuccess(i -> context.verify(() -> {
+              assertThat(recreated.get()).isTrue();
+              checkpoint.flag();
+            }))
+            .onFailure(context::failNow)
+        )
+        .onFailure(context::failNow);
+  }
+
+  @Test
+  @SuppressWarnings("unchecked")
+  public void shouldNotRecreateProducerWhenBootstrapServerNotChanged(
+      final VertxTestContext context) {
+
+    final RequestToRecordMapper<Object, Object> mapper
+        = (request) -> Future.succeededFuture();
+
+    final var first = new AtomicBoolean(true);
+    final var recreated = new AtomicBoolean(false);
+
+    final var handler = new RequestHandler<Object, Object>(
+        new Properties(),
+        mapper,
+        properties -> {
+          if (!first.getAndSet(false)) {
+            context.failNow(new IllegalStateException("producer should be recreated"));
+          }
+          return mock(KafkaProducer.class);
+        }
+    );
+
+    final var checkpoint = context.checkpoint();
+
+    final var broker1 = new BrokerWrapper(Broker.newBuilder()
+        .setId("1")
+        .setBootstrapServers("kafka-1:9092,kafka-2:9092")
+        .build());
+
+    final var broker2 = new BrokerWrapper(Broker.newBuilder()
+        .setId("1")
+        .setBootstrapServers("kafka-1:9092,kafka-2:9092")
+        .build());
+
+    handler.reconcile(Map.of(broker1, new HashSet<>()))
+        .onSuccess(ignored -> handler.reconcile(Map.of(broker2, new HashSet<>()))
+            .onSuccess(i -> context.verify(() -> {
+              assertThat(recreated.get()).isFalse();
+              checkpoint.flag();
+            }))
+            .onFailure(context::failNow)
+        )
+        .onFailure(context::failNow);
+  }
 }
\ No newline at end of file
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/integration/ReceiverVerticleTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/integration/ReceiverVerticleTest.java
index b61d77292..572f6da72 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/integration/ReceiverVerticleTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/integration/ReceiverVerticleTest.java
@@ -47,6 +47,7 @@ import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
@@ -80,7 +81,11 @@ public class ReceiverVerticleTest {
         new CloudEventSerializer()
     );
     KafkaProducer<String, CloudEvent> producer = KafkaProducer.create(vertx, mockProducer);
-    handler = new RequestHandler<>(producer, new CloudEventRequestToRecordMapper());
+    handler = new RequestHandler<>(
+        new Properties(),
+        new CloudEventRequestToRecordMapper(),
+        properties -> producer
+    );
 
     final var httpServerOptions = new HttpServerOptions();
     httpServerOptions.setPort(PORT);
diff --git a/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java b/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
index 682db8073..9272baffd 100644
--- a/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
+++ b/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
@@ -61,7 +61,6 @@ import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
-import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.junit.jupiter.api.AfterAll;
@@ -167,6 +166,7 @@ public class DataPlaneTest {
                 .setTopic(TOPIC)
                 .setNamespace(BROKER_NAMESPACE)
                 .setName(BROKER_NAME)
+                .setBootstrapServers(bootstrapServers())
                 .setId(UUID.randomUUID().toString())
                 .build()),
             Set.of(
@@ -318,10 +318,11 @@ public class DataPlaneTest {
       final Vertx vertx,
       final VertxTestContext context) throws InterruptedException {
 
-    final Properties configs = producerConfigs();
-    final KafkaProducer<String, CloudEvent> producer = KafkaProducer.create(vertx, configs);
-
-    final var handler = new RequestHandler<>(producer, new CloudEventRequestToRecordMapper());
+    final var handler = new RequestHandler<>(
+        producerConfigs(),
+        new CloudEventRequestToRecordMapper(),
+        properties -> KafkaProducer.create(vertx, properties)
+    );
 
     final var httpServerOptions = new HttpServerOptions();
     httpServerOptions.setPort(INGRESS_PORT);
@@ -337,9 +338,12 @@ public class DataPlaneTest {
 
   private static Properties producerConfigs() {
     final var configs = new Properties();
-    configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, format("localhost:%d", KAFKA_PORT));
     configs.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
     configs.put(VALUE_SERIALIZER_CLASS_CONFIG, CloudEventSerializer.class);
     return configs;
   }
+
+  private static String bootstrapServers() {
+    return format("localhost:%d", KAFKA_PORT);
+  }
 }