Take the input time from the job, fixes tests

Change the DataAccessITest to use a single version of time
hawkular · Mar 29, 2018 · 871e94c · 871e94c
1 parent fe3001d
commit 871e94c
Show file tree

Hide file tree

Showing 6 changed files with 21 additions and 13 deletions.
diff --git a/core/metrics-core-service/src/main/java/org/hawkular/metrics/core/service/DataAccess.java b/core/metrics-core-service/src/main/java/org/hawkular/metrics/core/service/DataAccess.java
@@ -74,7 +74,7 @@ public interface DataAccess {
      */
 //    Completable resetTempTable(long timestamp);
 
-    Set<Long> findExpiredTables();
+    Set<Long> findExpiredTables(long startTime);
 
     Observable<Observable<Row>> findAllDataFromBucket(long timestamp, int pageSize, int maxConcurrency);
 

diff --git a/.../metrics-core-service/src/main/java/org/hawkular/metrics/core/service/DataAccessImpl.java b/.../metrics-core-service/src/main/java/org/hawkular/metrics/core/service/DataAccessImpl.java
@@ -807,10 +807,9 @@ public <T> Observable<Row> findMetricsInMetricsIndex(String tenantId, MetricType
     }
 
     @Override
-    public Set<Long> findExpiredTables() {
-        long currentTime = System.currentTimeMillis();
-        Long currentTableKey = prepMap.floorKey(currentTime);
-        return prepMap.subMap(0L, false, currentTableKey, false).keySet();
+    public Set<Long> findExpiredTables(long startTime) {
+        Long currentTableKey = prepMap.floorKey(startTime);
+        return prepMap.subMap(0L, false, currentTableKey, true).keySet();
     }
 
     /**

diff --git a/...rics-core-service/src/main/java/org/hawkular/metrics/core/service/MetricsServiceImpl.java b/...rics-core-service/src/main/java/org/hawkular/metrics/core/service/MetricsServiceImpl.java
@@ -776,7 +776,7 @@ public Completable verifyAndCreateTempTables(ZonedDateTime startTime, ZonedDateT
     @SuppressWarnings("unchecked")
     public Completable compressBlock(long jobStartTimeSlice, int pageSize, int maxConcurrency) {
         return Completable.fromObservable(
-                Observable.from(dataAccess.findExpiredTables())
+                Observable.from(dataAccess.findExpiredTables(jobStartTimeSlice))
                         .flatMap(startTimeSlice ->
                                 dataAccess.findAllDataFromBucket(startTimeSlice, pageSize, maxConcurrency)
                                         .switchIfEmpty(Observable.empty())

diff --git a/...metrics-core-service/src/test/java/org/hawkular/metrics/core/service/DataAccessITest.java b/...metrics-core-service/src/test/java/org/hawkular/metrics/core/service/DataAccessITest.java
@@ -31,6 +31,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.hawkular.metrics.core.service.transformers.MetricIdentifierFromFullDataRowTransformer;
+import org.hawkular.metrics.datetime.DateTimeService;
 import org.hawkular.metrics.model.AvailabilityType;
 import org.hawkular.metrics.model.DataPoint;
 import org.hawkular.metrics.model.Metric;
@@ -63,6 +64,7 @@ public class DataAccessITest extends BaseITest {
     private static int DEFAULT_PAGE_SIZE = 5000;
 
     private DataAccessImpl dataAccess;
+    private static final DateTime now = DateTimeService.now.get();
 
     private PreparedStatement truncateTenants;
     private PreparedStatement truncateGaugeData;
@@ -71,7 +73,7 @@ public class DataAccessITest extends BaseITest {
 
     @BeforeClass
     public void initClass() {
-        this.dataAccess = (DataAccessImpl) TestDataAccessFactory.newInstance(session);
+        this.dataAccess = (DataAccessImpl) TestDataAccessFactory.newInstance(session, now());
 
         truncateTenants = session.prepare("TRUNCATE tenants");
         truncateGaugeData = session.prepare("TRUNCATE data");
@@ -208,7 +210,7 @@ public void insertAndFindAvailabilities() throws Exception {
 
     @Test
     public void findAllMetricsPartitionKeys() throws Exception {
-        long start = now().getMillis();
+        long start = now.getMillis();
 
         Observable.from(asList(
                 new Metric<>(new MetricId<>("t1", GAUGE, "m1"), singletonList(new DataPoint<>(start, 0.1))),
@@ -229,6 +231,10 @@ public void findAllMetricsPartitionKeys() throws Exception {
         assertEquals(metrics.size(), 4);
     }
 
+    private static DateTime now() {
+        return new DateTime(now);
+    }
+
     @Test
     void testFindAllDataFromBucket() throws Exception {
         String tenantId = "t1";

diff --git a/...cs-core-service/src/test/java/org/hawkular/metrics/core/service/DelegatingDataAccess.java b/...cs-core-service/src/test/java/org/hawkular/metrics/core/service/DelegatingDataAccess.java
@@ -116,8 +116,8 @@ public <T> Observable<Row> findMetricsInMetricsIndex(String tenantId, MetricType
         return delegate.findMetricsInMetricsIndex(tenantId, type);
     }
 
-    @Override public Set<Long> findExpiredTables() {
-        return delegate.findExpiredTables();
+    @Override public Set<Long> findExpiredTables(long startTime) {
+        return delegate.findExpiredTables(startTime);
     }
 
     @Override public Observable<Observable<Row>> findAllDataFromBucket(long timestamp, int pageSize, int maxConcurrency) {

diff --git a/...s-core-service/src/test/java/org/hawkular/metrics/core/service/TestDataAccessFactory.java b/...s-core-service/src/test/java/org/hawkular/metrics/core/service/TestDataAccessFactory.java
@@ -40,6 +40,10 @@ public class TestDataAccessFactory {
     private static final CoreLogger log = CoreLogging.getCoreLogger(TestDataAccessFactory.class);
 
     public static DataAccess newInstance(Session session) {
+        return newInstance(session, DateTimeService.now.get());
+    }
+
+    public static DataAccess newInstance(Session session, DateTime now) {
         session.execute(String.format("USE %s", BaseITest.getKeyspace()));
         final CountDownLatch latch = new CountDownLatch(3);
         final CountDownLatch fallBackTable = new CountDownLatch(0);
@@ -55,7 +59,7 @@ void prepareTempStatements(String tableName, Long mapKey) {
                 }
             }
         };
-        dataAccess.createTempTablesIfNotExists(tableListForTesting())
+        dataAccess.createTempTablesIfNotExists(tableListForTesting(now))
                 .subscribeOn(Schedulers.io())
                 .toBlocking().subscribe();
         try {
@@ -70,9 +74,8 @@ void prepareTempStatements(String tableName, Long mapKey) {
     /**
      * Create few temporary tables for tests
      */
-    static Set<Long> tableListForTesting() {
+    static Set<Long> tableListForTesting(DateTime now) {
         Set<Long> tempTables = new HashSet<>(3);
-        DateTime now = DateTimeService.now.get();
         tempTables.add(now.getMillis());
         tempTables.add(now.minusHours(2).getMillis());
         tempTables.add(now.plusHours(2).getMillis());