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

Merge pull request #9591 from dashevchenko/cacheFix

Browse Source 
Transactional cache issue
This commit is contained in:
Andrew Shvayka 2023-11-16 16:06:22 +02:00 committed by GitHub
commit 36739e7469
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 145 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
common/cache/src/main/java/org/thingsboard/server/cache/TbTransactionalCache.java vendored
View File

@ -47,6 +47,18 @@ public interface TbTransactionalCache<K extends Serializable, V extends Serializ
*/
TbCacheTransaction<K, V> newTransactionForKeys(List<K> keys);
default V getOrFetchFromDB(K key, Supplier<V> dbCall, boolean cacheNullValue, boolean putToCache) {
if (putToCache) {
return getAndPutInTransaction(key, dbCall, cacheNullValue);
} else {
TbCacheValueWrapper<V> cacheValueWrapper = get(key);
if (cacheValueWrapper != null) {
return cacheValueWrapper.get();
}
return dbCall.get();
}
}
default V getAndPutInTransaction(K key, Supplier<V> dbCall, boolean cacheNullValue) {
TbCacheValueWrapper<V> cacheValueWrapper = get(key);
if (cacheValueWrapper != null) {
@ -69,6 +81,19 @@ public interface TbTransactionalCache<K extends Serializable, V extends Serializ
}
}
default <R> R getOrFetchFromDB(K key, Supplier<R> dbCall, Function<V, R> cacheValueToResult, Function<R, V> dbValueToCacheValue, boolean cacheNullValue, boolean putToCache) {
if (putToCache) {
return getAndPutInTransaction(key, dbCall, cacheValueToResult, dbValueToCacheValue, cacheNullValue);
} else {
TbCacheValueWrapper<V> cacheValueWrapper = get(key);
if (cacheValueWrapper != null) {
var cacheValue = cacheValueWrapper.get();
return cacheValue == null ? null : cacheValueToResult.apply(cacheValue);
}
return dbCall.get();
}
}
default <R> R getAndPutInTransaction(K key, Supplier<R> dbCall, Function<V, R> cacheValueToResult, Function<R, V> dbValueToCacheValue, boolean cacheNullValue) {
TbCacheValueWrapper<V> cacheValueWrapper = get(key);
if (cacheValueWrapper != null) {

4
common/dao-api/src/main/java/org/thingsboard/server/dao/asset/AssetProfileService.java
View File

@ -27,8 +27,12 @@ public interface AssetProfileService extends EntityDaoService {
AssetProfile findAssetProfileById(TenantId tenantId, AssetProfileId assetProfileId);
AssetProfile findAssetProfileById(TenantId tenantId, AssetProfileId assetProfileId, boolean putInCache);
AssetProfile findAssetProfileByName(TenantId tenantId, String profileName);
AssetProfile findAssetProfileByName(TenantId tenantId, String profileName, boolean putInCache);
AssetProfileInfo findAssetProfileInfoById(TenantId tenantId, AssetProfileId assetProfileId);
AssetProfile saveAssetProfile(AssetProfile assetProfile, boolean doValidate);

4
common/dao-api/src/main/java/org/thingsboard/server/dao/device/DeviceProfileService.java
View File

@ -27,8 +27,12 @@ public interface DeviceProfileService extends EntityDaoService {
DeviceProfile findDeviceProfileById(TenantId tenantId, DeviceProfileId deviceProfileId);
DeviceProfile findDeviceProfileById(TenantId tenantId, DeviceProfileId deviceProfileId, boolean putInCache);
DeviceProfile findDeviceProfileByName(TenantId tenantId, String profileName);
DeviceProfile findDeviceProfileByName(TenantId tenantId, String profileName, boolean putInCache);
DeviceProfileInfo findDeviceProfileInfoById(TenantId tenantId, DeviceProfileId deviceProfileId);
DeviceProfile saveDeviceProfile(DeviceProfile deviceProfile, boolean doValidate);

2
common/dao-api/src/main/java/org/thingsboard/server/dao/entityview/EntityViewService.java
View File

@ -50,6 +50,8 @@ public interface EntityViewService extends EntityDaoService {
EntityView findEntityViewById(TenantId tenantId, EntityViewId entityViewId);
EntityView findEntityViewById(TenantId tenantId, EntityViewId entityViewId, boolean putInCache);
EntityView findEntityViewByTenantIdAndName(TenantId tenantId, String name);
PageData<EntityView> findEntityViewByTenantId(TenantId tenantId, PageLink pageLink);

24
dao/src/main/java/org/thingsboard/server/dao/asset/AssetProfileServiceImpl.java
View File

@ -90,18 +90,28 @@ public class AssetProfileServiceImpl extends AbstractCachedEntityService<AssetPr
@Override
public AssetProfile findAssetProfileById(TenantId tenantId, AssetProfileId assetProfileId) {
return findAssetProfileById(tenantId, assetProfileId, true);
}
@Override
public AssetProfile findAssetProfileById(TenantId tenantId, AssetProfileId assetProfileId, boolean putInCache) {
log.trace("Executing findAssetProfileById [{}]", assetProfileId);
Validator.validateId(assetProfileId, INCORRECT_ASSET_PROFILE_ID + assetProfileId);
return cache.getAndPutInTransaction(AssetProfileCacheKey.fromId(assetProfileId),
() -> assetProfileDao.findById(tenantId, assetProfileId.getId()), true);
return cache.getOrFetchFromDB(AssetProfileCacheKey.fromId(assetProfileId),
() -> assetProfileDao.findById(tenantId, assetProfileId.getId()), true, putInCache);
}
@Override
public AssetProfile findAssetProfileByName(TenantId tenantId, String profileName) {
return findAssetProfileByName(tenantId, profileName, true);
}
@Override
public AssetProfile findAssetProfileByName(TenantId tenantId, String profileName, boolean putInCache) {
log.trace("Executing findAssetProfileByName [{}][{}]", tenantId, profileName);
Validator.validateString(profileName, INCORRECT_ASSET_PROFILE_NAME + profileName);
return cache.getAndPutInTransaction(AssetProfileCacheKey.fromName(tenantId, profileName),
() -> assetProfileDao.findByName(tenantId, profileName), false);
return cache.getOrFetchFromDB(AssetProfileCacheKey.fromName(tenantId, profileName),
() -> assetProfileDao.findByName(tenantId, profileName), false, putInCache);
}
@Override
@ -127,7 +137,7 @@ public class AssetProfileServiceImpl extends AbstractCachedEntityService<AssetPr
if (doValidate) {
oldAssetProfile = assetProfileValidator.validate(assetProfile, AssetProfile::getTenantId);
} else if (assetProfile.getId() != null) {
oldAssetProfile = findAssetProfileById(assetProfile.getTenantId(), assetProfile.getId());
oldAssetProfile = findAssetProfileById(assetProfile.getTenantId(), assetProfile.getId(), false);
}
AssetProfile savedAssetProfile;
try {
@ -208,13 +218,13 @@ public class AssetProfileServiceImpl extends AbstractCachedEntityService<AssetPr
@Override
public AssetProfile findOrCreateAssetProfile(TenantId tenantId, String name) {
log.trace("Executing findOrCreateAssetProfile");
AssetProfile assetProfile = findAssetProfileByName(tenantId, name);
AssetProfile assetProfile = findAssetProfileByName(tenantId, name, false);
if (assetProfile == null) {
try {
assetProfile = this.doCreateDefaultAssetProfile(tenantId, name, name.equals("default"));
} catch (DataValidationException e) {
if (ASSET_PROFILE_WITH_SUCH_NAME_ALREADY_EXISTS.equals(e.getMessage())) {
assetProfile = findAssetProfileByName(tenantId, name);
assetProfile = findAssetProfileByName(tenantId, name, false);
} else {
throw e;
}

24
dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java
View File

@ -113,18 +113,28 @@ public class DeviceProfileServiceImpl extends AbstractCachedEntityService<Device
@Override
public DeviceProfile findDeviceProfileById(TenantId tenantId, DeviceProfileId deviceProfileId) {
return findDeviceProfileById(tenantId, deviceProfileId, true);
}
@Override
public DeviceProfile findDeviceProfileById(TenantId tenantId, DeviceProfileId deviceProfileId, boolean putInCache) {
log.trace("Executing findDeviceProfileById [{}]", deviceProfileId);
validateId(deviceProfileId, INCORRECT_DEVICE_PROFILE_ID + deviceProfileId);
return cache.getAndPutInTransaction(DeviceProfileCacheKey.fromId(deviceProfileId),
() -> deviceProfileDao.findById(tenantId, deviceProfileId.getId()), true);
return cache.getOrFetchFromDB(DeviceProfileCacheKey.fromId(deviceProfileId),
() -> deviceProfileDao.findById(tenantId, deviceProfileId.getId()), true, putInCache);
}
@Override
public DeviceProfile findDeviceProfileByName(TenantId tenantId, String profileName) {
return findDeviceProfileByName(tenantId, profileName, true);
}
@Override
public DeviceProfile findDeviceProfileByName(TenantId tenantId, String profileName, boolean putInCache) {
log.trace("Executing findDeviceProfileByName [{}][{}]", tenantId, profileName);
validateString(profileName, INCORRECT_DEVICE_PROFILE_NAME + profileName);
return cache.getAndPutInTransaction(DeviceProfileCacheKey.fromName(tenantId, profileName),
() -> deviceProfileDao.findByName(tenantId, profileName), true);
return cache.getOrFetchFromDB(DeviceProfileCacheKey.fromName(tenantId, profileName),
() -> deviceProfileDao.findByName(tenantId, profileName), true, putInCache);
}
@Override
@ -164,7 +174,7 @@ public class DeviceProfileServiceImpl extends AbstractCachedEntityService<Device
if (doValidate) {
oldDeviceProfile = deviceProfileValidator.validate(deviceProfile, DeviceProfile::getTenantId);
} else if (deviceProfile.getId() != null) {
oldDeviceProfile = findDeviceProfileById(deviceProfile.getTenantId(), deviceProfile.getId());
oldDeviceProfile = findDeviceProfileById(deviceProfile.getTenantId(), deviceProfile.getId(), false);
}
DeviceProfile savedDeviceProfile;
try {
@ -252,13 +262,13 @@ public class DeviceProfileServiceImpl extends AbstractCachedEntityService<Device
@Override
public DeviceProfile findOrCreateDeviceProfile(TenantId tenantId, String name) {
log.trace("Executing findOrCreateDefaultDeviceProfile");
DeviceProfile deviceProfile = findDeviceProfileByName(tenantId, name);
DeviceProfile deviceProfile = findDeviceProfileByName(tenantId, name, false);
if (deviceProfile == null) {
try {
deviceProfile = this.doCreateDefaultDeviceProfile(tenantId, name, name.equals("default"));
} catch (DataValidationException e) {
if (DEVICE_PROFILE_WITH_SUCH_NAME_ALREADY_EXISTS.equals(e.getMessage())) {
deviceProfile = findDeviceProfileByName(tenantId, name);
deviceProfile = findDeviceProfileByName(tenantId, name, false);
} else {
throw e;
}

2
dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java
View File

@ -223,7 +223,7 @@ public class DeviceServiceImpl extends AbstractCachedEntityService<DeviceCacheKe
}
device.setDeviceProfileId(new DeviceProfileId(deviceProfile.getId().getId()));
} else {
deviceProfile = this.deviceProfileService.findDeviceProfileById(device.getTenantId(), device.getDeviceProfileId());
deviceProfile = this.deviceProfileService.findDeviceProfileById(device.getTenantId(), device.getDeviceProfileId(), false);
if (deviceProfile == null) {
throw new DataValidationException("Device is referencing non existing device profile!");
}

9
dao/src/main/java/org/thingsboard/server/dao/entityview/EntityViewServiceImpl.java
View File

@ -163,11 +163,16 @@ public class EntityViewServiceImpl extends AbstractCachedEntityService<EntityVie
@Override
public EntityView findEntityViewById(TenantId tenantId, EntityViewId entityViewId) {
return findEntityViewById(tenantId, entityViewId, true);
}
@Override
public EntityView findEntityViewById(TenantId tenantId, EntityViewId entityViewId, boolean putInCache) {
log.trace("Executing findEntityViewById [{}]", entityViewId);
validateId(entityViewId, INCORRECT_ENTITY_VIEW_ID + entityViewId);
return cache.getAndPutInTransaction(EntityViewCacheKey.byId(entityViewId),
return cache.getOrFetchFromDB(EntityViewCacheKey.byId(entityViewId),
() -> entityViewDao.findById(tenantId, entityViewId.getId())
, EntityViewCacheValue::getEntityView, v -> new EntityViewCacheValue(v, null), true);
, EntityViewCacheValue::getEntityView, v -> new EntityViewCacheValue(v, null), true, putInCache);
}
@Override

39
dao/src/test/java/org/thingsboard/server/dao/service/AssetServiceTest.java
View File

@ -20,16 +20,25 @@ import org.junit.Assert;
import org.junit.Test;
import org.junit.jupiter.api.Assertions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;
import org.thingsboard.server.common.data.Customer;
import org.thingsboard.server.common.data.EntitySubtype;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.Tenant;
import org.thingsboard.server.common.data.asset.Asset;
import org.thingsboard.server.common.data.asset.AssetInfo;
import org.thingsboard.server.common.data.asset.AssetProfile;
import org.thingsboard.server.common.data.id.AssetProfileId;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.dao.asset.AssetDao;
import org.thingsboard.server.dao.asset.AssetProfileService;
import org.thingsboard.server.dao.asset.AssetService;
import org.thingsboard.server.dao.customer.CustomerService;
import org.thingsboard.server.dao.exception.DataValidationException;
@ -37,6 +46,7 @@ import org.thingsboard.server.dao.exception.DataValidationException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID;
@ -46,7 +56,13 @@ public class AssetServiceTest extends AbstractServiceTest {
@Autowired
AssetService assetService;
@Autowired
AssetDao assetDao;
@Autowired
CustomerService customerService;
@Autowired
private AssetProfileService assetProfileService;
@Autowired
private PlatformTransactionManager platformTransactionManager;
private IdComparator<Asset> idComparator = new IdComparator<>();
@ -75,6 +91,29 @@ public class AssetServiceTest extends AbstractServiceTest {
assetService.deleteAsset(tenantId, savedAsset.getId());
}
@Test
public void testShouldNotPutInCacheRolledbackAssetProfile() {
AssetProfile assetProfile = new AssetProfile();
assetProfile.setName(StringUtils.randomAlphabetic(10));
assetProfile.setTenantId(tenantId);
Asset asset = new Asset();
asset.setName("My asset" + StringUtils.randomAlphabetic(15));
asset.setType(assetProfile.getName());
asset.setTenantId(tenantId);
DefaultTransactionDefinition def = new DefaultTransactionDefinition();
TransactionStatus status = platformTransactionManager.getTransaction(def);
try {
assetProfileService.saveAssetProfile(assetProfile);
assetService.saveAsset(asset);
} finally {
platformTransactionManager.rollback(status);
}
AssetProfile assetProfileByName = assetProfileService.findAssetProfileByName(tenantId, assetProfile.getName());
Assert.assertNull(assetProfileByName);
}
@Test
public void testSaveAssetWithEmptyName() {
Asset asset = new Asset();

29
dao/src/test/java/org/thingsboard/server/dao/service/DeviceServiceTest.java
View File

@ -22,6 +22,9 @@ import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.Assertions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.thingsboard.server.common.data.Customer;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.DeviceInfo;
@ -32,6 +35,8 @@ import org.thingsboard.server.common.data.OtaPackage;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.Tenant;
import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.asset.Asset;
import org.thingsboard.server.common.data.asset.AssetProfile;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.ota.ChecksumAlgorithm;
@ -72,6 +77,8 @@ public class DeviceServiceTest extends AbstractServiceTest {
OtaPackageService otaPackageService;
@Autowired
TenantProfileService tenantProfileService;
@Autowired
private PlatformTransactionManager platformTransactionManager;
private IdComparator<Device> idComparator = new IdComparator<>();
private TenantId anotherTenantId;
@ -305,6 +312,28 @@ public class DeviceServiceTest extends AbstractServiceTest {
});
}
@Test
public void testShouldNotPutInCacheRolledbackDeviceProfile() {
DeviceProfile deviceProfile = createDeviceProfile(tenantId, "New device Profile" + StringUtils.randomAlphabetic(5));
Device device = new Device();
device.setType(deviceProfile.getName());
device.setTenantId(tenantId);
device.setName("My device"+ StringUtils.randomAlphabetic(5));
DefaultTransactionDefinition def = new DefaultTransactionDefinition();
TransactionStatus status = platformTransactionManager.getTransaction(def);
try {
deviceProfileService.saveDeviceProfile(deviceProfile);
deviceService.saveDevice(device);
} finally {
platformTransactionManager.rollback(status);
}
DeviceProfile deviceProfileByName = deviceProfileService.findDeviceProfileByName(tenantId, deviceProfile.getName());
Assert.assertNull(deviceProfileByName);
}
@Test
public void testAssignDeviceToNonExistentCustomer() {
Device device = new Device();