Merge pull request #14 from svishevsky/feature/add-batch-size-limit-f…

…or-wal-completer Added batch size limit for WAL Completer
PlaytikaOSS · Feb 8, 2022 · c0665a7 · c0665a7
2 parents 74fb847 + 43513b0
commit c0665a7
Show file tree

Hide file tree

Showing 20 changed files with 212 additions and 69 deletions.
diff --git a/...ater/src/main/java/nosql/batch/update/aerospike/basic/wal/AerospikeBasicWalCompleter.java b/...ater/src/main/java/nosql/batch/update/aerospike/basic/wal/AerospikeBasicWalCompleter.java
@@ -17,14 +17,15 @@ public class AerospikeBasicWalCompleter {
 
     public static <LOCKS> WriteAheadLogCompleter<LOCKS, List<Record>, AerospikeLock, Value> basicCompleter(
             BatchOperations<LOCKS, List<Record>, AerospikeLock, Value> batchOperations,
-            Duration staleBatchesThreshold){
+            Duration staleBatchesThreshold, int batchSize){
         WriteAheadLogManager<LOCKS, List<Record>, Value> writeAheadLogManager
                 = batchOperations.getWriteAheadLogManager();
         AerospikeWriteAheadLogManager aerospikeWriteAheadLogManager = (AerospikeWriteAheadLogManager)writeAheadLogManager;
 
         return new WriteAheadLogCompleter<>(
                 batchOperations,
                 staleBatchesThreshold,
+                batchSize,
                 new AerospikeExclusiveLocker(
                         aerospikeWriteAheadLogManager.getClient(),
                         aerospikeWriteAheadLogManager.getWalNamespace(),

diff --git a/...updater/src/main/java/nosql/batch/update/aerospike/wal/AerospikeWriteAheadLogManager.java b/...updater/src/main/java/nosql/batch/update/aerospike/wal/AerospikeWriteAheadLogManager.java
@@ -16,6 +16,7 @@
 import nosql.batch.update.BatchUpdate;
 import nosql.batch.update.aerospike.lock.AerospikeBatchLocks;
 import nosql.batch.update.wal.WalRecord;
+import nosql.batch.update.wal.WalTimeRange;
 import nosql.batch.update.wal.WriteAheadLogManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -27,7 +28,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
-import java.util.concurrent.ExecutorService;
 
 public class AerospikeWriteAheadLogManager<LOCKS extends AerospikeBatchLocks<EV>, UPDATES, EV>
         implements WriteAheadLogManager<LOCKS, UPDATES, Value> {
@@ -101,10 +101,22 @@ public boolean deleteBatch(Value batchId) {
     }
 
     @Override
-    public List<WalRecord<LOCKS, UPDATES, Value>> getStaleBatches(Duration staleThreshold) {
+    public List<WalTimeRange> getTimeRanges(Duration staleThreshold, int batchSize) {
         Statement statement = staleBatchesStatement(staleThreshold, walNamespace, walSetName, clock);
         RecordSet recordSet = client.query(null, statement);
 
+        List<Long> timestamps = new ArrayList<>();
+        recordSet.iterator().forEachRemaining(keyRecord -> timestamps.add(keyRecord.record.getLong(TIMESTAMP_BIN_NAME)));
+        Collections.sort(timestamps);
+
+        return getTimeRangesForTimestamps(timestamps, batchSize);
+    }
+
+    @Override
+    public List<WalRecord<LOCKS, UPDATES, Value>> getStaleBatchesForRange(WalTimeRange timeRange) {
+        Statement statement = staleBatchesStatement(walNamespace, walSetName, timeRange.getFromTimestamp(), timeRange.getToTimestamp());
+        RecordSet recordSet = client.query(null, statement);
+
         List<WalRecord<LOCKS, UPDATES, Value>> staleTransactions = new ArrayList<>();
         recordSet.iterator().forEachRemaining(keyRecord -> {
             Record record = keyRecord.record;
@@ -127,6 +139,37 @@ public static Statement staleBatchesStatement(Duration staleThreshold, String wa
         return statement;
     }
 
+    public static Statement staleBatchesStatement(String walNamespace, String walSetName, long begin, long end) {
+        Statement statement = new Statement();
+        statement.setNamespace(walNamespace);
+        statement.setSetName(walSetName);
+        statement.setFilter(Filter.range(TIMESTAMP_BIN_NAME, begin, end));
+        return statement;
+    }
+
+    public static  List<WalTimeRange> getTimeRangesForTimestamps(List<Long> timestamps, int batchSize) {
+        List<WalTimeRange> walTimeRanges = new ArrayList<>();
+
+        int fromIdx = 0;
+        int size = timestamps.size();
+        int toIdx = Math.min(batchSize, size) - 1;
+
+        while (fromIdx < size) {
+            long fromTimestamp = timestamps.get(fromIdx);
+            long toTimestamp = timestamps.get(toIdx);
+            walTimeRanges.add(new WalTimeRange(fromTimestamp, toTimestamp));
+
+            fromIdx = toIdx;
+            while (fromIdx < size && timestamps.get(fromIdx) == toTimestamp) {
+                fromIdx++;
+            }
+
+            toIdx = Math.min(fromIdx + batchSize, size) - 1;
+        }
+
+        return walTimeRanges;
+    }
+
     static byte[] getBytesFromUUID(UUID uuid) {
         ByteBuffer bb = ByteBuffer.wrap(new byte[16]);
         bb.putLong(uuid.getMostSignificantBits());

diff --git a/...tch-updater/src/test/java/nosql/batch/update/aerospike/basic/BasicBatchRetentionTest.java b/...tch-updater/src/test/java/nosql/batch/update/aerospike/basic/BasicBatchRetentionTest.java
@@ -46,10 +46,11 @@ public class BasicBatchRetentionTest extends BatchRetentionTest {
             = new BatchUpdater<>(operations);
 
     public static final Duration STALE_BATCHES_THRESHOLD = Duration.ofSeconds(1);
+    public static final int BATCH_SIZE = 100;
 
     static WriteAheadLogCompleter<AerospikeBasicBatchLocks, List<Record>, AerospikeLock, Value> walCompleter
             = new WriteAheadLogCompleter<>(
-            operations, STALE_BATCHES_THRESHOLD,
+            operations, STALE_BATCHES_THRESHOLD, BATCH_SIZE,
             new BasicRecoveryTest.DummyExclusiveLocker(),
             Executors.newScheduledThreadPool(1));
 
@@ -72,7 +73,7 @@ protected void checkForConsistency() {
         assertThat(getValue(key1, client)).isEqualTo(getValue(key2, client));
 
         await().timeout(ONE_SECOND).untilAsserted(() ->
-                assertThat(operations.getWriteAheadLogManager().getStaleBatches(STALE_BATCHES_THRESHOLD)).isEmpty());
+                assertThat(operations.getWriteAheadLogManager().getTimeRanges(STALE_BATCHES_THRESHOLD, BATCH_SIZE)).isEmpty());
     }
 
     @Override

diff --git a/...ike-batch-updater/src/test/java/nosql/batch/update/aerospike/basic/BasicRecoveryTest.java b/...ike-batch-updater/src/test/java/nosql/batch/update/aerospike/basic/BasicRecoveryTest.java
@@ -48,10 +48,11 @@ public class BasicRecoveryTest extends RecoveryTest {
             = new BatchUpdater<>(operations);
 
     public static final Duration STALE_BATCHES_THRESHOLD = Duration.ofSeconds(1);
+    public static final int BATCH_SIZE = 100;
 
     static WriteAheadLogCompleter<AerospikeBasicBatchLocks, List<Record>, AerospikeLock, Value> walCompleter
             = new WriteAheadLogCompleter<>(
-            operations, STALE_BATCHES_THRESHOLD,
+            operations, STALE_BATCHES_THRESHOLD, BATCH_SIZE,
             new DummyExclusiveLocker(),
             Executors.newScheduledThreadPool(1));
 
@@ -78,7 +79,7 @@ protected void checkForConsistency() {
         assertThat(getValue(key1, client)).isEqualTo(getValue(key2, client));
 
         await().timeout(ONE_SECOND).untilAsserted(() ->
-                assertThat(operations.getWriteAheadLogManager().getStaleBatches(STALE_BATCHES_THRESHOLD)).isEmpty());
+                assertThat(operations.getWriteAheadLogManager().getTimeRanges(STALE_BATCHES_THRESHOLD, BATCH_SIZE)).isEmpty());
     }
 
     @Override

diff --git a/...ter/src/test/java/nosql/batch/update/aerospike/wal/AerospikeWriteAheadLogManagerTest.java b/...ter/src/test/java/nosql/batch/update/aerospike/wal/AerospikeWriteAheadLogManagerTest.java
@@ -11,6 +11,7 @@
 import org.testcontainers.containers.GenericContainer;
 
 import java.time.Duration;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -32,6 +33,7 @@ public class AerospikeWriteAheadLogManagerTest extends WriteAheadLogManagerTest<
         clock.setTime(1000);
     }
     static final Duration staleThreshold = Duration.ofMillis(100);
+    static final int batchSize = 100;
 
     static String walSetName = String.valueOf(AerospikeWriteAheadLogManagerTest.class.hashCode());
 
@@ -80,9 +82,11 @@ protected void switchClockAhead() {
 
     @Override
     protected List<Value> getStaleBatches() {
-        return writeAheadLogManager.getStaleBatches(staleThreshold).stream()
-                .map(record -> record.batchId)
-                .collect(Collectors.toList());
+        return writeAheadLogManager.getTimeRanges(staleThreshold, batchSize).stream()
+                                   .map(writeAheadLogManager::getStaleBatchesForRange)
+                                   .flatMap(Collection::stream)
+                                   .map(record -> record.batchId)
+                                   .collect(Collectors.toList());
     }
 
 }
diff --git a/.../main/java/nosql/batch/update/reactor/aerospike/basic/wal/AerospikeBasicWalCompleter.java b/.../main/java/nosql/batch/update/reactor/aerospike/basic/wal/AerospikeBasicWalCompleter.java
@@ -17,14 +17,15 @@ public class AerospikeBasicWalCompleter {
 
     public static <LOCKS> ReactorWriteAheadLogCompleter<LOCKS, List<Record>, AerospikeLock, Value> basicCompleter(
             ReactorBatchOperations<LOCKS, List<Record>, AerospikeLock, Value> batchOperations,
-            Duration staleBatchesThreshold){
+            Duration staleBatchesThreshold, int batchSize){
         ReactorWriteAheadLogManager<LOCKS, List<Record>, Value> writeAheadLogManager
                 = batchOperations.getWriteAheadLogManager();
         AerospikeReactorWriteAheadLogManager aerospikeReactorWriteAheadLogManager = (AerospikeReactorWriteAheadLogManager)writeAheadLogManager;
 
         return new ReactorWriteAheadLogCompleter<>(
                 batchOperations,
                 staleBatchesThreshold,
+                batchSize,
                 new AerospikeExclusiveLocker(
                         aerospikeReactorWriteAheadLogManager.getClient(),
                         aerospikeReactorWriteAheadLogManager.getWalNamespace(),

diff --git a/...n/java/nosql/batch/update/reactor/aerospike/wal/AerospikeReactorWriteAheadLogManager.java b/...n/java/nosql/batch/update/reactor/aerospike/wal/AerospikeReactorWriteAheadLogManager.java
@@ -9,7 +9,6 @@
 import com.aerospike.client.Value;
 import com.aerospike.client.policy.RecordExistsAction;
 import com.aerospike.client.policy.WritePolicy;
-import com.aerospike.client.query.Filter;
 import com.aerospike.client.query.IndexType;
 import com.aerospike.client.query.RecordSet;
 import com.aerospike.client.query.Statement;
@@ -19,6 +18,7 @@
 import nosql.batch.update.aerospike.wal.AerospikeBatchUpdateSerde;
 import nosql.batch.update.reactor.wal.ReactorWriteAheadLogManager;
 import nosql.batch.update.wal.WalRecord;
+import nosql.batch.update.wal.WalTimeRange;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Mono;
@@ -32,6 +32,7 @@
 import java.util.UUID;
 
 import static nosql.batch.update.aerospike.wal.AerospikeWriteAheadLogManager.generateBatchId;
+import static nosql.batch.update.aerospike.wal.AerospikeWriteAheadLogManager.getTimeRangesForTimestamps;
 import static nosql.batch.update.aerospike.wal.AerospikeWriteAheadLogManager.staleBatchesStatement;
 
 public class AerospikeReactorWriteAheadLogManager<LOCKS extends AerospikeBatchLocks<EV>, UPDATES, EV>
@@ -107,10 +108,22 @@ public Mono<Boolean> deleteBatch(Value batchId) {
     }
 
     @Override
-    public List<WalRecord<LOCKS, UPDATES, Value>> getStaleBatches(Duration staleThreshold) {
+    public List<WalTimeRange> getTimeRanges(Duration staleThreshold, int batchSize) {
         Statement statement = staleBatchesStatement(staleThreshold, walNamespace, walSetName, clock);
         RecordSet recordSet = client.query(null, statement);
 
+        List<Long> timestamps = new ArrayList<>();
+        recordSet.iterator().forEachRemaining(keyRecord -> timestamps.add(keyRecord.record.getLong(TIMESTAMP_BIN_NAME)));
+        Collections.sort(timestamps);
+
+        return getTimeRangesForTimestamps(timestamps, batchSize);
+    }
+
+    @Override
+    public List<WalRecord<LOCKS, UPDATES, Value>> getStaleBatchesForRange(WalTimeRange timeRange) {
+        Statement statement = staleBatchesStatement(walNamespace, walSetName, timeRange.getFromTimestamp(), timeRange.getToTimestamp());
+        RecordSet recordSet = client.query(null, statement);
+
         List<WalRecord<LOCKS, UPDATES, Value>> staleTransactions = new ArrayList<>();
         recordSet.iterator().forEachRemaining(keyRecord -> {
             Record record = keyRecord.record;

diff --git a/...ter/src/test/java/nosql/batch/update/reactor/aerospike/basic/BasicBatchRetentionTest.java b/...ter/src/test/java/nosql/batch/update/reactor/aerospike/basic/BasicBatchRetentionTest.java
@@ -51,10 +51,11 @@ public class BasicBatchRetentionTest extends BatchRetentionTest {
             = new ReactorBatchUpdater<>(operations);
 
     public static final Duration STALE_BATCHES_THRESHOLD = Duration.ofSeconds(1);
+    public static final int BATCH_SIZE = 100;
 
     static ReactorWriteAheadLogCompleter<AerospikeBasicBatchLocks, List<Record>, AerospikeLock, Value> walCompleter
             = new ReactorWriteAheadLogCompleter<>(
-            operations, STALE_BATCHES_THRESHOLD,
+            operations, STALE_BATCHES_THRESHOLD, BATCH_SIZE,
             new BasicRecoveryTest.DummyExclusiveLocker(),
             Executors.newScheduledThreadPool(1));
 
@@ -77,7 +78,7 @@ protected void checkForConsistency() {
         assertThat(getValue(key1, client)).isEqualTo(getValue(key2, client));
 
         await().timeout(ONE_SECOND).untilAsserted(() ->
-                assertThat(operations.getWriteAheadLogManager().getStaleBatches(STALE_BATCHES_THRESHOLD)).isEmpty());
+                assertThat(operations.getWriteAheadLogManager().getTimeRanges(STALE_BATCHES_THRESHOLD, BATCH_SIZE)).isEmpty());
     }
 
     @Override

diff --git a/...h-updater/src/test/java/nosql/batch/update/reactor/aerospike/basic/BasicRecoveryTest.java b/...h-updater/src/test/java/nosql/batch/update/reactor/aerospike/basic/BasicRecoveryTest.java
@@ -51,10 +51,11 @@ public class BasicRecoveryTest extends RecoveryTest {
             = new ReactorBatchUpdater<>(operations);
 
     public static final Duration STALE_BATCHES_THRESHOLD = Duration.ofSeconds(1);
+    public static final int BATCH_SIZE = 100;
 
     static ReactorWriteAheadLogCompleter<AerospikeBasicBatchLocks, List<Record>, AerospikeLock, Value> walCompleter
             = new ReactorWriteAheadLogCompleter<>(
-            operations, STALE_BATCHES_THRESHOLD,
+            operations, STALE_BATCHES_THRESHOLD, BATCH_SIZE,
             new DummyExclusiveLocker(),
             Executors.newScheduledThreadPool(1));
 
@@ -81,7 +82,7 @@ protected void checkForConsistency() {
         assertThat(getValue(key1, client)).isEqualTo(getValue(key2, client));
 
         await().timeout(ONE_SECOND).untilAsserted(() ->
-                assertThat(operations.getWriteAheadLogManager().getStaleBatches(STALE_BATCHES_THRESHOLD)).isEmpty());
+                assertThat(operations.getWriteAheadLogManager().getTimeRanges(STALE_BATCHES_THRESHOLD, BATCH_SIZE)).isEmpty());
     }
 
     @Override

diff --git a/...va/nosql/batch/update/reactor/aerospike/wal/AerospikeReactorWriteAheadLogManagerTest.java b/...va/nosql/batch/update/reactor/aerospike/wal/AerospikeReactorWriteAheadLogManagerTest.java
@@ -14,6 +14,7 @@
 import org.testcontainers.containers.GenericContainer;
 
 import java.time.Duration;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -36,6 +37,7 @@ public class AerospikeReactorWriteAheadLogManagerTest extends WriteAheadLogManag
         clock.setTime(1000);
     }
     static final Duration staleThreshold = Duration.ofMillis(100);
+    static final int batchSize = 100;
 
     static String walSetName = String.valueOf(AerospikeReactorWriteAheadLogManagerTest.class.hashCode());
 
@@ -84,7 +86,9 @@ protected void switchClockAhead() {
 
     @Override
     protected List<Value> getStaleBatches() {
-        return writeAheadLogManager.getStaleBatches(staleThreshold).stream()
+        return writeAheadLogManager.getTimeRanges(staleThreshold, batchSize).stream()
+                .map(writeAheadLogManager::getStaleBatchesForRange)
+                .flatMap(Collection::stream)
                 .map(record -> record.batchId)
                 .collect(Collectors.toList());
     }