Merge pull request #1008: [ingest-server] fix prefix scan without limit

Author: je-ik

direct/ingest-server/src/main/java/cz/o2/proxima/direct/server/RetrieveService.java

@@ -222,7 +222,7 @@ private static List<KeyValue<Object>> processListRequest(
       if (addedKvs.get() == 0 || failedPredicate.get()) {
         break;
       }
-    } while (limit > 0 && kvs.size() < limit);
+    } while (limit <= 0 || kvs.size() < limit);
     return kvs;
   }
 

direct/ingest-server/src/test/java/cz/o2/proxima/direct/server/RetrieveServiceTest.java

@@ -491,6 +491,13 @@ public void testListValid() {
     runListTestWithPrefix("wildcard.*");
   }
 
+  @Test
+  public void testListValidMany() {
+    runListTestWithPrefixMany("wildcard");
+    runListTestWithPrefixMany("wildcard.");
+    runListTestWithPrefixMany("wildcard.*");
+  }
+
   private void runListTestWithPrefix(String prefix) {
     EntityDescriptor entity = server.repo.getEntity("dummy");
     AttributeDescriptor<?> attribute = entity.getAttribute("wildcard.*");
@@ -560,6 +567,61 @@ public void onCompleted() {
     assertArrayEquals(new byte[] {1, 2, 3, 4}, response.getValue(1).getValue().toByteArray());
   }
 
+  private void runListTestWithPrefixMany(String prefix) {
+    EntityDescriptor entity = server.repo.getEntity("dummy");
+    AttributeDescriptor<?> attribute = entity.getAttribute("wildcard.*");
+    String key = "my-fancy-entity-key";
+
+    for (int i = 1; i <= 500; i++) {
+      Optionals.get(server.direct.getWriter(attribute))
+          .write(
+              StreamElement.upsert(
+                  entity,
+                  attribute,
+                  UUID.randomUUID().toString(),
+                  key,
+                  "wildcard." + i,
+                  System.currentTimeMillis(),
+                  new byte[] {1, 2, 3}),
+              CommitCallback.noop());
+    }
+    Rpc.ListRequest request =
+        Rpc.ListRequest.newBuilder()
+            .setEntity(entity.getName())
+            .setWildcardPrefix(prefix)
+            .setKey(key)
+            .build();
+
+    List<Rpc.ListResponse> responses = new ArrayList<>();
+    AtomicBoolean finished = new AtomicBoolean(false);
+    final StreamObserver<Rpc.ListResponse> responseObserver;
+    responseObserver =
+        new StreamObserver<>() {
+          @Override
+          public void onNext(Rpc.ListResponse res) {
+            responses.add(res);
+          }
+
+          @Override
+          public void onError(Throwable thrwbl) {
+            throw new RuntimeException(thrwbl);
+          }
+
+          @Override
+          public void onCompleted() {
+            finished.set(true);
+          }
+        };
+
+    retrieve.listAttributes(request, responseObserver);
+
+    assertTrue(finished.get());
+    assertEquals(1, responses.size());
+    Rpc.ListResponse response = responses.get(0);
+    assertEquals(200, response.getStatus());
+    assertEquals(500, response.getValueCount());
+  }
+
   @Test
   public void testMultifetchListValid() {
     EntityDescriptor entity = server.repo.getEntity("dummy");