From 2b212d777c67d1eb7b6a30011ae267cf29c47554 Mon Sep 17 00:00:00 2001 From: David Date: Thu, 16 Aug 2018 17:24:59 +0300 Subject: [PATCH] formatting ,added INDEFINITE to possible partitioning in yml ,savePartition returns immediateFuture(null) if FixedPartition mode enabled --- .../src/main/resources/thingsboard.yml | 2 +- .../dao/timeseries/BaseTimeseriesService.java | 8 +++--- .../CassandraBaseTimeseriesDao.java | 26 +++++++++---------- 3 files changed, 18 insertions(+), 18 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index f510ec48a5..4d7dc7f41e 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -201,7 +201,7 @@ cassandra: read_consistency_level: "${CASSANDRA_READ_CONSISTENCY_LEVEL:ONE}" write_consistency_level: "${CASSANDRA_WRITE_CONSISTENCY_LEVEL:ONE}" default_fetch_size: "${CASSANDRA_DEFAULT_FETCH_SIZE:2000}" - # Specify partitioning size for timestamp key-value storage. Example MINUTES, HOURS, DAYS, MONTHS + # Specify partitioning size for timestamp key-value storage. Example MINUTES, HOURS, DAYS, MONTHS,INDEFINITE ts_key_value_partitioning: "${TS_KV_PARTITIONING:MONTHS}" ts_key_value_ttl: "${TS_KV_TTL:0}" buffer_size: "${CASSANDRA_QUERY_BUFFER_SIZE:200000}" diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java index c981378939..6c00735c4d 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java @@ -1,12 +1,12 @@ /** * Copyright © 2016-2018 The Thingsboard Authors - * + *

* Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

* Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java index 591023443c..f25a8b4e63 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java @@ -1,12 +1,12 @@ /** * Copyright © 2016-2018 The Thingsboard Authors - * + *

* Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

* Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. @@ -163,18 +163,18 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem } } - public boolean isFixedPartitioning(){ - return tsFormat.getTruncateUnit().equals(TsPartitionDate.FIXED_PARTITION); + public boolean isFixedPartitioning() { + return tsFormat.getTruncateUnit().equals(TsPartitionDate.FIXED_PARTITION); } - private ListenableFuture> getPartitionsFuture(TsKvQuery query, EntityId entityId, long minPartition, long maxPartition){ + private ListenableFuture> getPartitionsFuture(TsKvQuery query, EntityId entityId, long minPartition, long maxPartition) { - if(isFixedPartitioning()){ //no need to fetch partitions from DB + if (isFixedPartitioning()) { //no need to fetch partitions from DB return Futures.immediateFuture(FIXED_PARTITION); } ResultSetFuture partitionsFuture = fetchPartitions(entityId, query.getKey(), minPartition, maxPartition); - return Futures.transform(partitionsFuture, getPartitionsArrayFunction(), readResultsProcessingExecutor); + return Futures.transform(partitionsFuture, getPartitionsArrayFunction(), readResultsProcessingExecutor); } @@ -183,7 +183,7 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem long minPartition = toPartitionTs(query.getStartTs()); long maxPartition = toPartitionTs(query.getEndTs()); - final ListenableFuture> partitionsListFuture = getPartitionsFuture(query,entityId,minPartition,maxPartition); + final ListenableFuture> partitionsListFuture = getPartitionsFuture(query, entityId, minPartition, maxPartition); final SimpleListenableFuture> resultFuture = new SimpleListenableFuture<>(); Futures.addCallback(partitionsListFuture, new FutureCallback>() { @@ -244,7 +244,7 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem final long ts = startTs + (endTs - startTs) / 2; - ListenableFuture> partitionsListFuture = getPartitionsFuture(query,entityId, minPartition, maxPartition); + ListenableFuture> partitionsListFuture = getPartitionsFuture(query, entityId, minPartition, maxPartition); ListenableFuture> aggregationChunks = Futures.transformAsync(partitionsListFuture, getFetchChunksAsyncFunction(entityId, key, aggregation, startTs, endTs), readResultsProcessingExecutor); @@ -320,8 +320,8 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem @Override public ListenableFuture savePartition(EntityId entityId, long tsKvEntryTs, String key, long ttl) { - if(isFixedPartitioning()){ - return getFuture(null, rs -> null); + if (isFixedPartitioning()) { + return Futures.immediateFuture(null); } ttl = computeTtl(ttl); long partition = toPartitionTs(tsKvEntryTs);