run update-codegen

Signed-off-by: debasishbsws <debasishbsws.dev@gmail.com>
knative-extensions · Aug 22, 2023 · 3b92a3b · 3b92a3b
1 parent 0bf6b9b
commit 3b92a3b
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 17 deletions.
diff --git a/...a/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java b/...a/dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/OrderedConsumerVerticle.java
@@ -139,14 +139,17 @@ private void poll() {
         if (this.isPollInFlight.compareAndSet(false, true)) {
             logger.debug("Polling for records {}", getConsumerVerticleContext().getLoggingKeyValue());
 
-            this.consumer.poll(POLLING_TIMEOUT).onSuccess(records -> vertx.runOnContext(v -> this.recordsHandler(records))).onFailure(t -> {
-                if (this.closed.get()) {
-                    // The failure might have been caused by stopping the consumer, so we just ignore it
-                    return;
-                }
-                isPollInFlight.set(false);
-                exceptionHandler(t);
-            });
+            this.consumer
+                    .poll(POLLING_TIMEOUT)
+                    .onSuccess(records -> vertx.runOnContext(v -> this.recordsHandler(records)))
+                    .onFailure(t -> {
+                        if (this.closed.get()) {
+                            // The failure might have been caused by stopping the consumer, so we just ignore it
+                            return;
+                        }
+                        isPollInFlight.set(false);
+                        exceptionHandler(t);
+                    });
         }
     }
 

diff --git a/...dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticle.java b/...dev/knative/eventing/kafka/broker/dispatcher/impl/consumer/UnorderedConsumerVerticle.java
@@ -97,15 +97,18 @@ private synchronized void poll() {
             return;
         }
         if (this.isPollInFlight.compareAndSet(false, true)) {
-            this.consumer.poll(POLL_TIMEOUT).onSuccess(records -> vertx.runOnContext(v -> this.handleRecords(records))).onFailure(cause -> {
-                isPollInFlight.set(false);
-                logger.error(
-                        "Failed to poll messages {}",
-                        getConsumerVerticleContext().getLoggingKeyValue(),
-                        cause);
-                // Wait before retrying.
-                vertx.setTimer(BACKOFF_DELAY_MS, t -> poll());
-            });
+            this.consumer
+                    .poll(POLL_TIMEOUT)
+                    .onSuccess(records -> vertx.runOnContext(v -> this.handleRecords(records)))
+                    .onFailure(cause -> {
+                        isPollInFlight.set(false);
+                        logger.error(
+                                "Failed to poll messages {}",
+                                getConsumerVerticleContext().getLoggingKeyValue(),
+                                cause);
+                        // Wait before retrying.
+                        vertx.setTimer(BACKOFF_DELAY_MS, t -> poll());
+                    });
         }
     }