Repository: knative-sandbox/eventing-kafka-broker
Pull request #2461 (2022-08-03 10:12:30)

[release-1.6] Add TODO for max pool size override
This is an automated cherry-pick of #2453


Patch:
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index 2e3a7b2d3..c3536ce35 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -126,7 +126,7 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, ks *sources.Kafk
 						DeliverySpec: &eventingduck.DeliverySpec{
 							Retry:         pointer.Int32(10),
 							BackoffPolicy: &backoffPolicy,
-							BackoffDelay:  pointer.String("PT10S"),
+							BackoffDelay:  pointer.String("PT0.3S"),
 							Timeout:       pointer.String("PT600S"),
 						},
 						Ordering: DefaultDeliveryOrder,
diff --git a/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml b/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml
index 2289ab368..f3cbd9989 100644
--- a/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml
+++ b/data-plane/config/broker/100-config-kafka-broker-data-plane.yaml
@@ -115,7 +115,7 @@ data:
     value.deserializer=io.cloudevents.kafka.CloudEventDeserializer
     fetch.min.bytes=1
     heartbeat.interval.ms=3000
-    max.partition.fetch.bytes=1048576
+    max.partition.fetch.bytes=65536
     session.timeout.ms=10000
     # ssl.key.password=
     # ssl.keystore.location=
@@ -132,8 +132,8 @@ data:
     fetch.max.bytes=52428800
     isolation.level=read_uncommitted
     max.poll.interval.ms=300000
-    max.poll.records=500
-    # partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
+    max.poll.records=50
+    partition.assignment.strategy=org.apache.kafka.clients.consumer.StickyAssignor
     receive.buffer.bytes=65536
     request.timeout.ms=2000
     # sasl.client.callback.handler.class=
diff --git a/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml b/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
index ab8568576..1216f2b20 100644
--- a/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
+++ b/data-plane/config/channel/100-config-kafka-channel-data-plane.yaml
@@ -115,7 +115,7 @@ data:
     value.deserializer=io.cloudevents.kafka.CloudEventDeserializer
     fetch.min.bytes=1
     heartbeat.interval.ms=3000
-    max.partition.fetch.bytes=1048576
+    max.partition.fetch.bytes=65536
     session.timeout.ms=10000
     # ssl.key.password=
     # ssl.keystore.location=
@@ -131,8 +131,8 @@ data:
     fetch.max.bytes=52428800
     isolation.level=read_uncommitted
     max.poll.interval.ms=300000
-    max.poll.records=500
-    # partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
+    max.poll.records=50
+    partition.assignment.strategy=org.apache.kafka.clients.consumer.StickyAssignor
     receive.buffer.bytes=65536
     request.timeout.ms=2000
     # sasl.client.callback.handler.class=
diff --git a/data-plane/config/source/100-config-kafka-source-data-plane.yaml b/data-plane/config/source/100-config-kafka-source-data-plane.yaml
index 0cf1d99ad..27a4f8ef1 100644
--- a/data-plane/config/source/100-config-kafka-source-data-plane.yaml
+++ b/data-plane/config/source/100-config-kafka-source-data-plane.yaml
@@ -97,7 +97,7 @@ data:
     value.deserializer=io.cloudevents.kafka.CloudEventDeserializer
     fetch.min.bytes=1
     heartbeat.interval.ms=3000
-    max.partition.fetch.bytes=1048576
+    max.partition.fetch.bytes=65536
     session.timeout.ms=10000
     # ssl.key.password=
     # ssl.keystore.location=
@@ -114,8 +114,8 @@ data:
     fetch.max.bytes=52428800
     isolation.level=read_uncommitted
     max.poll.interval.ms=300000
-    max.poll.records=500
-    # partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
+    max.poll.records=50
+    partition.assignment.strategy=org.apache.kafka.clients.consumer.StickyAssignor
     receive.buffer.bytes=65536
     request.timeout.ms=2000
     # sasl.client.callback.handler.class=
diff --git a/data-plane/dispatcher/pom.xml b/data-plane/dispatcher/pom.xml
index 45fb03c53..7af5eebeb 100644
--- a/data-plane/dispatcher/pom.xml
+++ b/data-plane/dispatcher/pom.xml
@@ -156,6 +156,12 @@
                     <include>**</include>
                   </includes>
                 </filter>
+                <filter>
+                  <artifact>org.apache.kafka:kafka-clients</artifact>
+                  <includes>
+                    <include>**</include>
+                  </includes>
+                </filter>
               </filters>
             </configuration>
           </execution>
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
index af406db39..0bcc355c6 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
@@ -340,10 +340,11 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
                                                       final EgressConfig egressConfig,
                                                       final DataPlaneContract.Egress egress) {
     var webClientOptions = this.webClientOptions;
-    if (egress.getVReplicas() > 0) {
-      webClientOptions = new WebClientOptions(this.webClientOptions);
-      webClientOptions.setMaxPoolSize(egress.getVReplicas());
-    }
+    // TODO use numPartitions for ordered delivery or max.poll.records for unordered delivery
+    // if (egress.getVReplicas() > 0) {
+    //   webClientOptions = new WebClientOptions(this.webClientOptions);
+    //   webClientOptions.setMaxPoolSize(egress.getVReplicas());
+    // }
     return new WebClientCloudEventSender(vertx, WebClient.create(vertx, webClientOptions), target, egressConfig);
   }
 
diff --git a/data-plane/receiver/pom.xml b/data-plane/receiver/pom.xml
index bd737b3f6..5a0fe553f 100644
--- a/data-plane/receiver/pom.xml
+++ b/data-plane/receiver/pom.xml
@@ -148,6 +148,12 @@
                     <include>**</include>
                   </includes>
                 </filter>
+                <filter>
+                  <artifact>org.apache.kafka:kafka-clients</artifact>
+                  <includes>
+                    <include>**</include>
+                  </includes>
+                </filter>
               </filters>
             </configuration>
           </execution>


Test patch:
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 09d333442..2ace88d4a 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -112,7 +112,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -158,7 +158,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -286,7 +286,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -398,7 +398,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -447,7 +447,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -495,7 +495,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -527,7 +527,7 @@ func TestReconcileKind(t *testing.T) {
 									internals.Ordered,
 									NewConsumerTimeout("PT600S"),
 									NewConsumerRetry(10),
-									NewConsumerBackoffDelay("PT10S"),
+									NewConsumerBackoffDelay("PT0.3S"),
 									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 								),
 							),
@@ -585,7 +585,7 @@ func TestReconcileKind(t *testing.T) {
 									internals.Ordered,
 									NewConsumerTimeout("PT600S"),
 									NewConsumerRetry(10),
-									NewConsumerBackoffDelay("PT10S"),
+									NewConsumerBackoffDelay("PT0.3S"),
 									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 								),
 							),
@@ -642,7 +642,7 @@ func TestReconcileKind(t *testing.T) {
 									internals.Ordered,
 									NewConsumerTimeout("PT600S"),
 									NewConsumerRetry(10),
-									NewConsumerBackoffDelay("PT10S"),
+									NewConsumerBackoffDelay("PT0.3S"),
 									NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 								),
 							),
@@ -685,7 +685,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -729,7 +729,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -773,7 +773,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
@@ -818,7 +818,7 @@ func TestReconcileKind(t *testing.T) {
 								internals.Ordered,
 								NewConsumerTimeout("PT600S"),
 								NewConsumerRetry(10),
-								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffDelay("PT0.3S"),
 								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
 							),
 						),
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
index eecda4cca..9f549995c 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherTest.java
@@ -422,9 +422,8 @@ public class RecordDispatcherTest {
           .summaries()
           .stream()
           .reduce((a, b) -> b)
-          .get()
-          .max()
-      ).isGreaterThan(0)
+          .isPresent()
+      ).isTrue()
     );
   }