DLPX-94056 HOP: Performance Tuning causing Unit Test case to break (#8)

PR URL: https://www.github.com/delphix/hop/pull/8

Author: AnoopKumarDelphix

engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java

@@ -1579,23 +1579,11 @@ private Object[] handleGetRow() throws HopException {
       // The buffer to grow beyond "a few" entries.
       // We'll only do that if the previous transform has not ended...
 
-      if (!isStopped() && row == null) {
-        boolean streamReady = false;
-        // Check each other input stream to see that stream meets the threshold
-        for (int r = 0; r < inputRowSets.size(); r++) {
-          if (inputRowSet.isDone() || inputRowSet.size() > lowerBufferBoundary) {
-            streamReady = true;
-            break;
-          }
-          nextInputStream();
-          inputRowSet = currentInputStream();
-        }
-        if (!streamReady) {
-          try {
-            Thread.sleep(0, 1); // Minimum sleeps vary by OS scheduler, this could be 1ms or more
-          } catch (InterruptedException e) {
-            // Ignore sleep interruption exception
-          }
+      if (!inputRowSet.isDone() && inputRowSet.size() <= lowerBufferBoundary && !isStopped()) {
+        try {
+          Thread.sleep(0, 1);
+        } catch (InterruptedException e) {
+          // Ignore sleep interruption exception
         }
       }
 
@@ -1938,23 +1926,11 @@ public Object[] handleGetRowFrom(IRowSet rowSet) throws HopTransformException {
     // The buffer to grow beyond "a few" entries.
     // We'll only do that if the previous transform has not ended...
 
-    if (!isStopped()) {
-      boolean streamReady = false;
-      // Check each other input stream to see that stream meets the threshold
-      for (int r = 0; r < inputRowSets.size(); r++) {
-        if (rowSet.isDone() || rowSet.size() > lowerBufferBoundary) {
-          streamReady = true;
-          break;
-        }
-        nextInputStream();
-        rowSet = currentInputStream();
-      }
-      if (!streamReady) {
-        try {
-          Thread.sleep(0, 1); // Minimum sleeps vary by OS scheduler, this could be 1ms or more
-        } catch (InterruptedException e) {
-          // Ignore sleep interruption exception
-        }
+    if (!rowSet.isDone() && rowSet.size() <= lowerBufferBoundary && !isStopped()) {
+      try {
+        Thread.sleep(0, 1);
+      } catch (InterruptedException e) {
+        // Ignore sleep interruption exception
       }
     }
 

engine/src/test/java/org/apache/hop/pipeline/transform/DynamicWaitTimesTest.java

@@ -34,21 +34,21 @@ public class DynamicWaitTimesTest extends TestCase {
 
   public void testSingleStreamStatus() {
     IRowSet rowSet = new BlockingRowSet(3);
-    status = DynamicWaitTimes.build(Collections.singletonList(rowSet), () -> 0);
+    status = DynamicWaitTimes.build(Collections.singletonList(rowSet), () -> 0, 20);
     assertEquals(1, status.get());
     status.adjust(true, rowSet);
     assertEquals(2, status.get());
     for (int i = 0; i < 10; i++) {
       status.adjust(true, rowSet);
     }
-    assertEquals(DynamicWaitTimes.MAX_TIMEOUT, status.get());
+    assertEquals(20, status.get());
   }
 
   public void testMultiStreamStatus() {
     List<IRowSet> rowSetList =
         new ArrayList<>(
             Arrays.asList(new BlockingRowSet(1), new BlockingRowSet(2), new BlockingRowSet(7)));
-    status = DynamicWaitTimes.build(rowSetList, () -> activeStreamIndex.get());
+    status = DynamicWaitTimes.build(rowSetList, () -> activeStreamIndex.get(), 20);
     for (IRowSet iRowSet : rowSetList) {
       status.adjust(false, iRowSet);
       assertEquals(1, status.get());