Kafka cloud events

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

@@ -65,8 +65,6 @@
 import org.thingsboard.server.common.data.alarm.AlarmInfo;
 import org.thingsboard.server.common.data.asset.Asset;
 import org.thingsboard.server.common.data.asset.AssetInfo;
-import org.thingsboard.server.common.data.audit.ActionType;
-import org.thingsboard.server.common.data.cloud.CloudEventType;
 import org.thingsboard.server.common.data.asset.AssetProfile;
 import org.thingsboard.server.common.data.audit.ActionType;
 import org.thingsboard.server.common.data.domain.Domain;
@@ -127,9 +125,9 @@
 import org.thingsboard.server.dao.attributes.AttributesService;
 import org.thingsboard.server.dao.audit.AuditLogService;
 import org.thingsboard.server.dao.cloud.CloudEventService;
+import org.thingsboard.server.dao.cloud.EdgeSettingsService;
 import org.thingsboard.server.dao.customer.CustomerService;
 import org.thingsboard.server.dao.dashboard.DashboardService;
-import org.thingsboard.server.dao.device.ClaimDevicesService;
 import org.thingsboard.server.dao.device.DeviceCredentialsService;
 import org.thingsboard.server.dao.device.DeviceProfileService;
 import org.thingsboard.server.dao.device.DeviceService;
@@ -165,16 +163,12 @@
 import org.thingsboard.server.service.component.ComponentDiscoveryService;
 import org.thingsboard.server.service.entitiy.TbLogEntityActionService;
 import org.thingsboard.server.service.entitiy.user.TbUserSettingsService;
-import org.thingsboard.server.service.ota.OtaPackageStateService;
-import org.thingsboard.server.service.profile.TbAssetProfileCache;
 import org.thingsboard.server.service.profile.TbDeviceProfileCache;
 import org.thingsboard.server.service.security.model.SecurityUser;
 import org.thingsboard.server.service.security.permission.AccessControlService;
 import org.thingsboard.server.service.security.permission.Operation;
 import org.thingsboard.server.service.security.permission.Resource;
-import org.thingsboard.server.service.state.DeviceStateService;
 import org.thingsboard.server.service.sync.ie.exporting.ExportableEntitiesService;
-import org.thingsboard.server.service.sync.vc.EntitiesVersionControlService;
 import org.thingsboard.server.service.telemetry.AlarmSubscriptionService;
 import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
 
@@ -283,9 +277,6 @@ public abstract class BaseController {
     @Autowired
     protected AuditLogService auditLogService;
 
-    @Autowired
-    protected DeviceStateService deviceStateService;
-
     @Autowired
     protected EntityViewService entityViewService;
 
@@ -296,10 +287,10 @@ public abstract class BaseController {
     protected AttributesService attributesService;
 
     @Autowired
-    protected ClaimDevicesService claimDevicesService;
+    protected CloudEventService cloudEventService;
 
     @Autowired
-    protected CloudEventService cloudEventService;
+    protected EdgeSettingsService edgeSettingsService;
 
     @Autowired
     protected PartitionService partitionService;
@@ -310,9 +301,6 @@ public abstract class BaseController {
     @Autowired
     protected OtaPackageService otaPackageService;
 
-    @Autowired
-    protected OtaPackageStateService otaPackageStateService;
-
     @Autowired
     protected RpcService rpcService;
 
@@ -325,9 +313,6 @@ public abstract class BaseController {
     @Autowired
     protected TbDeviceProfileCache deviceProfileCache;
 
-    @Autowired
-    protected TbAssetProfileCache assetProfileCache;
-
     @Autowired(required = false)
     protected EdgeService edgeService;
 
@@ -340,9 +325,6 @@ public abstract class BaseController {
     @Autowired
     protected QueueService queueService;
 
-    @Autowired
-    protected EntitiesVersionControlService vcService;
-
     @Autowired
     protected ExportableEntitiesService entitiesService;
 
@@ -357,6 +339,10 @@ public abstract class BaseController {
     @Getter
     protected boolean edgesEnabled;
 
+    @Value("${queue.type}")
+    @Getter
+    protected String queueType;
+
     @ExceptionHandler(Exception.class)
     public void handleControllerException(Exception e, HttpServletResponse response) {
         ThingsboardException thingsboardException = handleException(e);

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

@@ -56,7 +56,6 @@
 import org.thingsboard.server.common.data.sync.ie.importing.csv.BulkImportRequest;
 import org.thingsboard.server.common.data.sync.ie.importing.csv.BulkImportResult;
 import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse;
-import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest;
 import org.thingsboard.server.config.annotations.ApiOperation;
 import org.thingsboard.server.dao.exception.DataValidationException;
 import org.thingsboard.server.dao.exception.IncorrectParameterException;
@@ -74,7 +73,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
-import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
@@ -473,7 +471,7 @@ public List<EntitySubtype> getEdgeTypes() throws ThingsboardException, Execution
     @PreAuthorize("hasAuthority('TENANT_ADMIN')")
     @PostMapping(value = "/edge/sync/{edgeId}")
     public DeferredResult<ResponseEntity> syncEdge(@Parameter(description = EDGE_ID_PARAM_DESCRIPTION, required = true)
-                         @PathVariable("edgeId") String strEdgeId) throws ThingsboardException {
+                                                   @PathVariable("edgeId") String strEdgeId) throws ThingsboardException {
         checkParameter("edgeId", strEdgeId);
         final DeferredResult<ResponseEntity> response = new DeferredResult<>();
         if (isEdgesEnabled() && edgeRpcServiceOpt.isPresent()) {
@@ -565,7 +563,7 @@ public EdgeSettings getEdgeSettings() throws ThingsboardException {
         try {
             SecurityUser user = getCurrentUser();
             TenantId tenantId = user.getTenantId();
-            return checkNotNull(cloudEventService.findEdgeSettings(tenantId));
+            return checkNotNull(edgeSettingsService.findEdgeSettings(tenantId));
         } catch (Exception e) {
             throw handleException(e);
         }
@@ -581,6 +579,10 @@ public PageData<CloudEvent> getCloudEvents(
             @RequestParam(required = false) String sortOrder,
             @RequestParam(required = false) Long startTime,
             @RequestParam(required = false) Long endTime) throws ThingsboardException {
+        if (queueType.equals("kafka")) {
+            throw new UnsupportedOperationException("getCloudEvents not supported for queue type - kafka");
+        }
+
         try {
             TenantId tenantId = getCurrentUser().getTenantId();
             TimePageLink pageLink = createTimePageLink(pageSize, page, textSearch, sortProperty, sortOrder, startTime, endTime);
@@ -600,6 +602,9 @@ public PageData<CloudEvent> getTimeseriesCloudEvents(
             @RequestParam(required = false) String sortOrder,
             @RequestParam(required = false) Long startTime,
             @RequestParam(required = false) Long endTime) throws ThingsboardException {
+        if (queueType.equals("kafka")) {
+            throw new UnsupportedOperationException("getTimeseriesCloudEvents not supported for queue type - kafka");
+        }
         try {
             TenantId tenantId = getCurrentUser().getTenantId();
             TimePageLink pageLink = createTimePageLink(pageSize, page, textSearch, sortProperty, sortOrder, startTime, endTime);

application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java

@@ -53,6 +53,9 @@ public class ThingsboardInstallService {
     @Value("${state.persistToTelemetry:false}")
     private boolean persistToTelemetry;
 
+    @Value("${queue.type}")
+    private String queueType;
+
     @Autowired
     private EntityDatabaseSchemaService entityDatabaseSchemaService;