dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Partitions should not be removed by custom TTL

Browse Source
This commit is contained in:
Volodymyr Babak 2021-07-30 14:36:42 +03:00 committed by Andrew Shvayka
parent e27ef59eed
commit d5319c9de0
6 changed files with 11 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java
View File

@ -100,7 +100,7 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq
}
@Override
public ListenableFuture<Integer> savePartition(TenantId tenantId, EntityId entityId, long tsKvEntryTs, String key, long ttl) {
public ListenableFuture<Integer> savePartition(TenantId tenantId, EntityId entityId, long tsKvEntryTs, String key) {
return Futures.immediateFuture(null);
}

2
dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java
View File

@ -124,7 +124,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
}
@Override
public ListenableFuture<Integer> savePartition(TenantId tenantId, EntityId entityId, long tsKvEntryTs, String key, long ttl) {
public ListenableFuture<Integer> savePartition(TenantId tenantId, EntityId entityId, long tsKvEntryTs, String key) {
return Futures.immediateFuture(0);
}

2
dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java
View File

@ -170,7 +170,7 @@ public class BaseTimeseriesService implements TimeseriesService {
if (entityId.getEntityType().equals(EntityType.ENTITY_VIEW)) {
throw new IncorrectParameterException("Telemetry data can't be stored for entity view. Read only");
}
futures.add(timeseriesDao.savePartition(tenantId, entityId, tsKvEntry.getTs(), tsKvEntry.getKey(), ttl));
futures.add(timeseriesDao.savePartition(tenantId, entityId, tsKvEntry.getTs(), tsKvEntry.getKey()));
futures.add(Futures.transform(timeseriesLatestDao.saveLatest(tenantId, entityId, tsKvEntry), v -> 0, MoreExecutors.directExecutor()));
futures.add(timeseriesDao.save(tenantId, entityId, tsKvEntry, ttl));
}

7
dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java
View File

@ -181,11 +181,14 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD
}
@Override
public ListenableFuture<Integer> savePartition(TenantId tenantId, EntityId entityId, long tsKvEntryTs, String key, long ttl) {
public ListenableFuture<Integer> savePartition(TenantId tenantId, EntityId entityId, long tsKvEntryTs, String key) {
if (isFixedPartitioning()) {
return Futures.immediateFuture(null);
}
ttl = computeTtl(ttl);
// DO NOT apply custom to partition, otherwise short TTL will remove partition too early
// partitions must remain in the DB forever or be removed only by systemTtl
// removal of empty partition is too expensive (we need to scan all data keys for this partitions with ALLOW FILTERING)
long ttl = computeTtl(0);
long partition = toPartitionTs(tsKvEntryTs);
if (cassandraTsPartitionsCache == null) {
return doSavePartition(tenantId, entityId, key, ttl, partition);

2
dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesDao.java
View File

@ -33,7 +33,7 @@ public interface TimeseriesDao {
ListenableFuture<Integer> save(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry, long ttl);
ListenableFuture<Integer> savePartition(TenantId tenantId, EntityId entityId, long tsKvEntryTs, String key, long ttl);
ListenableFuture<Integer> savePartition(TenantId tenantId, EntityId entityId, long tsKvEntryTs, String key);
ListenableFuture<Void> remove(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query);

4
dao/src/test/java/org/thingsboard/server/dao/nosql/CassandraPartitionsCacheTest.java
View File

@ -100,10 +100,10 @@ public class CassandraPartitionsCacheTest {
long tsKvEntryTs = System.currentTimeMillis();
for (int i = 0; i < 50000; i++) {
cassandraBaseTimeseriesDao.savePartition(tenantId, tenantId, tsKvEntryTs, "test" + i, 0);
cassandraBaseTimeseriesDao.savePartition(tenantId, tenantId, tsKvEntryTs, "test" + i);
}
for (int i = 0; i < 60000; i++) {
cassandraBaseTimeseriesDao.savePartition(tenantId, tenantId, tsKvEntryTs, "test" + i, 0);
cassandraBaseTimeseriesDao.savePartition(tenantId, tenantId, tsKvEntryTs, "test" + i);
}
verify(cassandraBaseTimeseriesDao, times(60000)).executeAsyncWrite(any(TenantId.class), any(Statement.class));
}