Making isLeaseTableEmpty use one paginated scan call. (#75)

Co-authored-by: Joshua Kim <kimjos@amazon.com>

src/main/java/com/amazonaws/services/kinesis/leases/impl/LeaseManager.java

@@ -240,7 +240,7 @@ public class LeaseManager<T extends Lease> implements ILeaseManager<T> {
      */
     @Override
     public boolean isLeaseTableEmpty() throws DependencyException, InvalidStateException, ProvisionedThroughputException {
-        return list(1).isEmpty();
+        return list(1, 1).isEmpty();
     }
 
     /**
@@ -253,6 +253,20 @@ public class LeaseManager<T extends Lease> implements ILeaseManager<T> {
      * @throws ProvisionedThroughputException if DynamoDB scan fail due to exceeded capacity
      */
     List<T> list(Integer limit) throws DependencyException, InvalidStateException, ProvisionedThroughputException {
+        return list(limit, Integer.MAX_VALUE);
+    }
+
+    /**
+     * List with the given page size, up to a limit of paginated calls.
+     *
+     * @param limit number of items to consider at a time - used by integration tests to force paging.
+     * @param maxPages max number of paginated scan calls.
+     * @return list of leases
+     * @throws InvalidStateException if table does not exist
+     * @throws DependencyException if DynamoDB scan fail in an unexpected way
+     * @throws ProvisionedThroughputException if DynamoDB scan fail due to exceeded capacity
+     */
+    private List<T> list(Integer limit, Integer maxPages) throws InvalidStateException, ProvisionedThroughputException, DependencyException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("Listing leases from table " + table);
         }
@@ -277,7 +291,7 @@ public class LeaseManager<T extends Lease> implements ILeaseManager<T> {
                 }
 
                 Map<String, AttributeValue> lastEvaluatedKey = scanResult.getLastEvaluatedKey();
-                if (lastEvaluatedKey == null) {
+                if (lastEvaluatedKey == null || --maxPages <= 0) {
                     // Signify that we're done.
                     scanResult = null;
                     if (LOG.isDebugEnabled()) {