fewer max outstanding records

responsivedev · Jun 26, 2024 · b5ef488 · b5ef488
1 parent 96a4daa
commit b5ef488
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 6 deletions.
diff --git a/...client-examples/e2e-test/src/main/java/dev/responsive/examples/e2etest/E2ETestDriver.java b/...client-examples/e2e-test/src/main/java/dev/responsive/examples/e2etest/E2ETestDriver.java
@@ -169,11 +169,14 @@ private void runProducer() {
     while (true) {
       synchronized (produceWait) {
         outstanding += produced;
-        while (outstanding >= maxOutstanding && keepRunning) {
-          try {
-            produceWait.wait();
-          } catch (final InterruptedException e) {
-            throw new RuntimeException(e);
+        if (outstanding > maxOutstanding) {
+          // wait for the produced records to drain
+          while (outstanding > 0 && keepRunning) {
+            try {
+              produceWait.wait();
+            } catch (final InterruptedException e) {
+              throw new RuntimeException(e);
+            }
           }
         }
       }

diff --git a/kafka-client-examples/e2e-test/src/main/java/dev/responsive/examples/e2etest/Params.java b/kafka-client-examples/e2e-test/src/main/java/dev/responsive/examples/e2etest/Params.java
@@ -11,7 +11,7 @@ public class Params {
   public static final int NUM_KEYS
       = Integer.parseInt(System.getenv().getOrDefault("NUM_KEYS", "16"));
   public static final int MAX_OUTSTANDING
-      = Integer.parseInt(System.getenv().getOrDefault("MAX_OUTSTANDING", "10000"));
+      = Integer.parseInt(System.getenv().getOrDefault("MAX_OUTSTANDING", "500"));
   public static final int RECEIVE_THRESHOLD
       = Integer.parseInt(System.getenv().getOrDefault("RECEIVE_THRESHOLD", "360"));
   public static final int FAULT_STOP_THRESHOLD