Refactor delete for latest and history

application/src/main/java/org/thingsboard/server/controller/TelemetryController.java

@@ -49,6 +49,7 @@ import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.rule.engine.api.AttributesDeleteRequest;
 import org.thingsboard.rule.engine.api.AttributesSaveRequest;
+import org.thingsboard.rule.engine.api.TimeseriesDeleteRequest;
 import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
 import org.thingsboard.server.common.adaptor.JsonConverter;
 import org.thingsboard.server.common.data.AttributeScope;
@@ -521,19 +522,25 @@ public class TelemetryController extends BaseController {
             for (String key : keys) {
                 deleteTsKvQueries.add(new BaseDeleteTsKvQuery(key, deleteFromTs, deleteToTs, rewriteLatestIfDeleted, deleteLatest));
             }
-            tsSubService.deleteTimeseriesAndNotify(tenantId, entityId, keys, deleteTsKvQueries, new FutureCallback<>() {
+            tsSubService.deleteTimeseries(TimeseriesDeleteRequest.builder()
-                @Override
+                    .tenantId(tenantId)
-                public void onSuccess(@Nullable Void tmp) {
+                    .entityId(entityId)
-                    logTimeseriesDeleted(user, entityId, keys, deleteFromTs, deleteToTs, null);
+                    .keys(keys)
-                    result.setResult(new ResponseEntity<>(HttpStatus.OK));
+                    .deleteHistoryQueries(deleteTsKvQueries)
-                }
+                    .callback(new FutureCallback<>() {
+                        @Override
+                        public void onSuccess(@Nullable List<String> tmp) {
+                            logTimeseriesDeleted(user, entityId, keys, deleteFromTs, deleteToTs, null);
+                            result.setResult(new ResponseEntity<>(HttpStatus.OK));
+                        }
 
-                @Override
+                        @Override
-                public void onFailure(Throwable t) {
+                        public void onFailure(Throwable t) {
-                    logTimeseriesDeleted(user, entityId, keys, deleteFromTs, deleteToTs, t);
+                            logTimeseriesDeleted(user, entityId, keys, deleteFromTs, deleteToTs, t);
-                    result.setResult(new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR));
+                            result.setResult(new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR));
-                }
+                        }
-            });
+                    })
+                    .build());
         });
     }
 

application/src/main/java/org/thingsboard/server/service/entitiy/entityview/DefaultTbEntityViewService.java

@@ -27,6 +27,7 @@ import org.springframework.stereotype.Service;
 import org.springframework.util.ConcurrentReferenceHashMap;
 import org.thingsboard.rule.engine.api.AttributesDeleteRequest;
 import org.thingsboard.rule.engine.api.AttributesSaveRequest;
+import org.thingsboard.rule.engine.api.TimeseriesDeleteRequest;
 import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
 import org.thingsboard.server.common.data.AttributeScope;
 import org.thingsboard.server.common.data.Customer;
@@ -58,6 +59,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
@@ -403,51 +405,32 @@ public class DefaultTbEntityViewService extends AbstractTbEntityService implemen
     private ListenableFuture<Void> deleteLatestFromEntityView(EntityView entityView, List<String> keys, User user) {
         EntityViewId entityId = entityView.getId();
         SettableFuture<Void> resultFuture = SettableFuture.create();
-        if (keys != null && !keys.isEmpty()) {
+        tsSubService.deleteTimeseries(TimeseriesDeleteRequest.builder()
-            tsSubService.deleteLatest(entityView.getTenantId(), entityId, keys, new FutureCallback<Void>() {
+                .tenantId(entityView.getTenantId())
-                @Override
+                .entityId(entityId)
-                public void onSuccess(@Nullable Void tmp) {
+                .keys(keys)
-                    try {
+                .callback(new FutureCallback<>() {
-                        logTimeseriesDeleted(entityView.getTenantId(), user, entityId, keys, null);
+                    @Override
-                    } catch (ThingsboardException e) {
+                    public void onSuccess(@Nullable List<String> result) {
-                        log.error("Failed to log timeseries delete", e);
+                        try {
-                    }
+                            logTimeseriesDeleted(entityView.getTenantId(), user, entityId, result, null);
-                    resultFuture.set(tmp);
+                        } catch (ThingsboardException e) {
-                }
+                            log.error("Failed to log timeseries delete", e);
-
+                        }
-                @Override
+                        resultFuture.set(null);
-                public void onFailure(Throwable t) {
-                    try {
-                        logTimeseriesDeleted(entityView.getTenantId(), user, entityId, keys, t);
-                    } catch (ThingsboardException e) {
-                        log.error("Failed to log timeseries delete", e);
-                    }
-                    resultFuture.setException(t);
-                }
-            });
-        } else {
-            tsSubService.deleteAllLatest(entityView.getTenantId(), entityId, new FutureCallback<Collection<String>>() {
-                @Override
-                public void onSuccess(@Nullable Collection<String> keys) {
-                    try {
-                        logTimeseriesDeleted(entityView.getTenantId(), user, entityId, new ArrayList<>(keys), null);
-                    } catch (ThingsboardException e) {
-                        log.error("Failed to log timeseries delete", e);
                     }
-                    resultFuture.set(null);
-                }
 
-                @Override
+                    @Override
-                public void onFailure(Throwable t) {
+                    public void onFailure(Throwable t) {
-                    try {
+                        try {
-                        logTimeseriesDeleted(entityView.getTenantId(), user, entityId, Collections.emptyList(), t);
+                            logTimeseriesDeleted(entityView.getTenantId(), user, entityId, Optional.ofNullable(keys).orElse(Collections.emptyList()), t);
-                    } catch (ThingsboardException e) {
+                        } catch (ThingsboardException e) {
-                        log.error("Failed to log timeseries delete", e);
+                            log.error("Failed to log timeseries delete", e);
+                        }
+                        resultFuture.setException(t);
                     }
-                    resultFuture.setException(t);
+                })
-                }
+                .build());
-            });
-        }
         return resultFuture;
     }
 

application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java

@@ -23,13 +23,16 @@ import jakarta.annotation.Nullable;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.DonAsynchron;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.rule.engine.api.AttributesDeleteRequest;
 import org.thingsboard.rule.engine.api.AttributesSaveRequest;
 import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
+import org.thingsboard.rule.engine.api.TimeseriesDeleteRequest;
 import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
 import org.thingsboard.server.common.data.ApiUsageRecordKey;
 import org.thingsboard.server.common.data.EntityType;
@@ -38,7 +41,6 @@ import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
-import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.data.kv.TsKvLatestRemovingResult;
 import org.thingsboard.server.common.msg.queue.TbCallback;
@@ -51,7 +53,6 @@ import org.thingsboard.server.service.entitiy.entityview.TbEntityViewService;
 import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
@@ -60,6 +61,7 @@ import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.function.Consumer;
 
 /**
  * Created by ashvayka on 27.03.18.
@@ -177,38 +179,26 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
     }
 
     @Override
-    public void deleteLatest(TenantId tenantId, EntityId entityId, List<String> keys, FutureCallback<Void> callback) {
+    public void deleteTimeseries(TimeseriesDeleteRequest request) {
-        checkInternalEntity(entityId);
+        checkInternalEntity(request.getEntityId());
-        deleteLatestInternal(tenantId, entityId, keys, callback);
+        deleteTimeseriesInternal(request);
-    }
-
-    @Override
-    public void deleteLatestInternal(TenantId tenantId, EntityId entityId, List<String> keys, FutureCallback<Void> callback) {
-        ListenableFuture<List<TsKvLatestRemovingResult>> deleteFuture = tsService.removeLatest(tenantId, entityId, keys);
-        addMainCallback(deleteFuture, callback);
     }
 
     @Override
-    public void deleteAllLatest(TenantId tenantId, EntityId entityId, FutureCallback<Collection<String>> callback) {
+    public void deleteTimeseriesInternal(TimeseriesDeleteRequest request) {
-        ListenableFuture<Collection<String>> deleteFuture = tsService.removeAllLatest(tenantId, entityId);
+        if (CollectionUtils.isNotEmpty(request.getKeys())) {
-        Futures.addCallback(deleteFuture, new FutureCallback<Collection<String>>() {
+            ListenableFuture<List<TsKvLatestRemovingResult>> deleteFuture;
-            @Override
+            if (request.getDeleteHistoryQueries() == null) {
-            public void onSuccess(@Nullable Collection<String> result) {
+                deleteFuture = tsService.removeLatest(request.getTenantId(), request.getEntityId(), request.getKeys());
-                callback.onSuccess(result);
+            } else {
+                deleteFuture = tsService.remove(request.getTenantId(), request.getEntityId(), request.getDeleteHistoryQueries());
+                addWsCallback(deleteFuture, result -> onTimeSeriesDelete(request.getTenantId(), request.getEntityId(), request.getKeys(), result));
             }
-
+            addMainCallback(deleteFuture, __ -> request.getCallback().onSuccess(request.getKeys()), request.getCallback()::onFailure);
-            @Override
+        } else {
-            public void onFailure(Throwable t) {
+            ListenableFuture<List<String>> deleteFuture = tsService.removeAllLatest(request.getTenantId(), request.getEntityId());
-                callback.onFailure(t);
+            addMainCallback(deleteFuture, request.getCallback()::onSuccess, request.getCallback()::onFailure);
-            }
+        }
-        }, tsCallBackExecutor);
-    }
-
-    @Override
-    public void deleteTimeseriesAndNotify(TenantId tenantId, EntityId entityId, List<String> keys, List<DeleteTsKvQuery> deleteTsKvQueries, FutureCallback<Void> callback) {
-        ListenableFuture<List<TsKvLatestRemovingResult>> deleteFuture = tsService.remove(tenantId, entityId, deleteTsKvQueries);
-        addMainCallback(deleteFuture, callback);
-        addWsCallback(deleteFuture, list -> onTimeSeriesDelete(tenantId, entityId, keys, list));
     }
 
     private void addEntityViewCallback(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts) {
@@ -314,17 +304,11 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer
 
     private <S> void addMainCallback(ListenableFuture<S> saveFuture, final FutureCallback<Void> callback) {
         if (callback == null) return;
-        Futures.addCallback(saveFuture, new FutureCallback<S>() {
+        addMainCallback(saveFuture, result -> callback.onSuccess(null), callback::onFailure);
-            @Override
+    }
-            public void onSuccess(@Nullable S result) {
-                callback.onSuccess(null);
-            }
 
-            @Override
+    private <S> void addMainCallback(ListenableFuture<S> saveFuture, Consumer<S> onSuccess, Consumer<Throwable> onFailure) {
-            public void onFailure(Throwable t) {
+        DonAsynchron.withCallback(saveFuture, onSuccess, onFailure, tsCallBackExecutor);
-                callback.onFailure(t);
-            }
-        }, tsCallBackExecutor);
     }
 
     private void checkInternalEntity(EntityId entityId) {

application/src/main/java/org/thingsboard/server/service/telemetry/InternalTelemetryService.java

@@ -15,16 +15,12 @@
  */
 package org.thingsboard.server.service.telemetry;
 
-import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.ListenableFuture;
 import org.thingsboard.rule.engine.api.AttributesDeleteRequest;
 import org.thingsboard.rule.engine.api.AttributesSaveRequest;
 import org.thingsboard.rule.engine.api.RuleEngineTelemetryService;
+import org.thingsboard.rule.engine.api.TimeseriesDeleteRequest;
 import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
-import org.thingsboard.server.common.data.id.EntityId;
-import org.thingsboard.server.common.data.id.TenantId;
-
-import java.util.List;
 
 /**
  * Created by ashvayka on 27.03.18.
@@ -35,8 +31,8 @@ public interface InternalTelemetryService extends RuleEngineTelemetryService {
 
     void saveAttributesInternal(AttributesSaveRequest request);
 
-    void deleteAttributesInternal(AttributesDeleteRequest request);
+    void deleteTimeseriesInternal(TimeseriesDeleteRequest request);
 
-    void deleteLatestInternal(TenantId tenantId, EntityId entityId, List<String> keys, FutureCallback<Void> callback);
+    void deleteAttributesInternal(AttributesDeleteRequest request);
 
 }

common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java

@@ -56,7 +56,7 @@ public interface TimeseriesService {
 
     ListenableFuture<List<TsKvLatestRemovingResult>> removeLatest(TenantId tenantId, EntityId entityId, Collection<String> keys);
 
-    ListenableFuture<Collection<String>> removeAllLatest(TenantId tenantId, EntityId entityId);
+    ListenableFuture<List<String>> removeAllLatest(TenantId tenantId, EntityId entityId);
 
     List<String> findAllKeysByDeviceProfileId(TenantId tenantId, DeviceProfileId deviceProfileId);
 

dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java

@@ -254,11 +254,11 @@ public class BaseTimeseriesService implements TimeseriesService {
     }
 
     @Override
-    public ListenableFuture<Collection<String>> removeAllLatest(TenantId tenantId, EntityId entityId) {
+    public ListenableFuture<List<String>> removeAllLatest(TenantId tenantId, EntityId entityId) {
         validate(entityId);
         return Futures.transformAsync(this.findAllLatest(tenantId, entityId), latest -> {
             if (latest != null && !latest.isEmpty()) {
-                Collection<String> keys = latest.stream().map(TsKvEntry::getKey).collect(Collectors.toList());
+                List<String> keys = latest.stream().map(TsKvEntry::getKey).collect(Collectors.toList());
                 return Futures.transform(this.removeLatest(tenantId, entityId, keys), res -> keys, MoreExecutors.directExecutor());
             } else {
                 return Futures.immediateFuture(Collections.emptyList());

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java

@@ -15,14 +15,6 @@
  */
 package org.thingsboard.rule.engine.api;
 
-import com.google.common.util.concurrent.FutureCallback;
-import org.thingsboard.server.common.data.id.EntityId;
-import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
-
-import java.util.Collection;
-import java.util.List;
-
 /**
  * Created by ashvayka on 02.04.18.
  */
@@ -32,12 +24,8 @@ public interface RuleEngineTelemetryService {
 
     void saveAttributes(AttributesSaveRequest request);
 
-    void deleteAttributes(AttributesDeleteRequest request);
+    void deleteTimeseries(TimeseriesDeleteRequest request);
-
-    void deleteLatest(TenantId tenantId, EntityId entityId, List<String> keys, FutureCallback<Void> callback);
-
-    void deleteAllLatest(TenantId tenantId, EntityId entityId, FutureCallback<Collection<String>> callback);
 
-    void deleteTimeseriesAndNotify(TenantId tenantId, EntityId entityId, List<String> keys, List<DeleteTsKvQuery> deleteTsKvQueries, FutureCallback<Void> callback);
+    void deleteAttributes(AttributesDeleteRequest request);
 
 }

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TimeseriesDeleteRequest.java

@@ -0,0 +1,85 @@
+/**
+ * Copyright © 2016-2024 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
+ *
+ * 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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.api;
+
+import com.google.common.util.concurrent.FutureCallback;
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.ToString;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
+
+import java.util.List;
+
+@Getter
+@ToString
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+public class TimeseriesDeleteRequest {
+
+    private final TenantId tenantId;
+    private final EntityId entityId;
+    private final List<String> keys;
+    private final List<DeleteTsKvQuery> deleteHistoryQueries;
+    private final FutureCallback<List<String>> callback;
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public static class Builder {
+
+        private TenantId tenantId;
+        private EntityId entityId;
+        private List<String> keys;
+        private List<DeleteTsKvQuery> deleteHistoryQueries;
+        private FutureCallback<List<String>> callback;
+
+        Builder() {}
+
+        public Builder tenantId(TenantId tenantId) {
+            this.tenantId = tenantId;
+            return this;
+        }
+
+        public Builder entityId(EntityId entityId) {
+            this.entityId = entityId;
+            return this;
+        }
+
+        public Builder keys(List<String> keys) {
+            this.keys = keys;
+            return this;
+        }
+
+        public Builder deleteHistoryQueries(List<DeleteTsKvQuery> deleteHistoryQueries) {
+            this.deleteHistoryQueries = deleteHistoryQueries;
+            return this;
+        }
+
+        public Builder callback(FutureCallback<List<String>> callback) {
+            this.callback = callback;
+            return this;
+        }
+
+        public TimeseriesDeleteRequest build() {
+            return new TimeseriesDeleteRequest(tenantId, entityId, keys, deleteHistoryQueries, callback);
+        }
+
+    }
+
+}